Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

22 files changed, 3858 insertions, 0 deletions

diff --git a/src/pybind/mgr/volumes/fs/operations/__init__.py b/src/pybind/mgr/volumes/fs/operations/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/__init__.py
diff --git a/src/pybind/mgr/volumes/fs/operations/access.py b/src/pybind/mgr/volumes/fs/operations/access.py
new file mode 100644
index 000000000..158e21c26
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/access.py
@@ -0,0 +1,139 @@
+import errno
+import json
+from typing import List
+
+def prepare_updated_caps_list(existing_caps, mds_cap_str, osd_cap_str, authorize=True):
+    caps_list = [] # type: List[str]
+    for k, v in existing_caps['caps'].items():
+        if k == 'mds' or k == 'osd':
+            continue
+        elif k == 'mon':
+            if not authorize and v == 'allow r':
+                continue
+        caps_list.extend((k,v))
+
+    if mds_cap_str:
+        caps_list.extend(('mds', mds_cap_str))
+    if osd_cap_str:
+        caps_list.extend(('osd', osd_cap_str))
+
+    if authorize and 'mon' not in caps_list:
+        caps_list.extend(('mon', 'allow r'))
+
+    return caps_list
+
+
+def allow_access(mgr, client_entity, want_mds_cap, want_osd_cap,
+                 unwanted_mds_cap, unwanted_osd_cap, existing_caps):
+    if existing_caps is None:
+        ret, out, err = mgr.mon_command({
+            "prefix": "auth get-or-create",
+            "entity": client_entity,
+            "caps": ['mds',  want_mds_cap, 'osd', want_osd_cap, 'mon', 'allow r'],
+            "format": "json"})
+    else:
+        cap = existing_caps[0]
+
+        def cap_update(
+                orig_mds_caps, orig_osd_caps, want_mds_cap,
+                want_osd_cap, unwanted_mds_cap, unwanted_osd_cap):
+
+            if not orig_mds_caps:
+                return want_mds_cap, want_osd_cap
+
+            mds_cap_tokens = [x.strip() for x in orig_mds_caps.split(",")]
+            osd_cap_tokens = [x.strip() for x in orig_osd_caps.split(",")]
+
+            if want_mds_cap in mds_cap_tokens:
+                return orig_mds_caps, orig_osd_caps
+
+            if unwanted_mds_cap in mds_cap_tokens:
+                mds_cap_tokens.remove(unwanted_mds_cap)
+                osd_cap_tokens.remove(unwanted_osd_cap)
+
+            mds_cap_tokens.append(want_mds_cap)
+            osd_cap_tokens.append(want_osd_cap)
+
+            return ",".join(mds_cap_tokens), ",".join(osd_cap_tokens)
+
+        orig_mds_caps = cap['caps'].get('mds', "")
+        orig_osd_caps = cap['caps'].get('osd', "")
+
+        mds_cap_str, osd_cap_str = cap_update(
+            orig_mds_caps, orig_osd_caps, want_mds_cap, want_osd_cap,
+            unwanted_mds_cap, unwanted_osd_cap)
+
+        caps_list = prepare_updated_caps_list(cap, mds_cap_str, osd_cap_str)
+        mgr.mon_command(
+            {
+                "prefix": "auth caps",
+                'entity': client_entity,
+                'caps': caps_list
+            })
+        ret, out, err = mgr.mon_command(
+            {
+                'prefix': 'auth get',
+                'entity': client_entity,
+                'format': 'json'
+            })
+
+    # Result expected like this:
+    # [
+    #     {
+    #         "entity": "client.foobar",
+    #         "key": "AQBY0\/pViX\/wBBAAUpPs9swy7rey1qPhzmDVGQ==",
+    #         "caps": {
+    #             "mds": "allow *",
+    #             "mon": "allow *"
+    #         }
+    #     }
+    # ]
+
+    caps = json.loads(out)
+    assert len(caps) == 1
+    assert caps[0]['entity'] == client_entity
+    return caps[0]['key']
+
+def deny_access(mgr, client_entity, want_mds_caps, want_osd_caps):
+    ret, out, err = mgr.mon_command({
+        "prefix": "auth get",
+        "entity": client_entity,
+        "format": "json",
+    })
+
+    if ret == -errno.ENOENT:
+        # Already gone, great.
+        return
+
+    def cap_remove(orig_mds_caps, orig_osd_caps, want_mds_caps, want_osd_caps):
+        mds_cap_tokens = [x.strip() for x in orig_mds_caps.split(",")]
+        osd_cap_tokens = [x.strip() for x in orig_osd_caps.split(",")]
+
+        for want_mds_cap, want_osd_cap in zip(want_mds_caps, want_osd_caps):
+            if want_mds_cap in mds_cap_tokens:
+                mds_cap_tokens.remove(want_mds_cap)
+                osd_cap_tokens.remove(want_osd_cap)
+                break
+
+        return ",".join(mds_cap_tokens), ",".join(osd_cap_tokens)
+
+    cap = json.loads(out)[0]
+    orig_mds_caps = cap['caps'].get('mds', "")
+    orig_osd_caps = cap['caps'].get('osd', "")
+    mds_cap_str, osd_cap_str = cap_remove(orig_mds_caps, orig_osd_caps,
+                                          want_mds_caps, want_osd_caps)
+
+    caps_list = prepare_updated_caps_list(cap, mds_cap_str, osd_cap_str, authorize=False)
+    if not caps_list:
+        mgr.mon_command(
+            {
+                'prefix': 'auth rm',
+                'entity': client_entity
+            })
+    else:
+        mgr.mon_command(
+            {
+                "prefix": "auth caps",
+                'entity': client_entity,
+                'caps': caps_list
+            })
diff --git a/src/pybind/mgr/volumes/fs/operations/clone_index.py b/src/pybind/mgr/volumes/fs/operations/clone_index.py
new file mode 100644
index 000000000..a2b31f858
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/clone_index.py
@@ -0,0 +1,98 @@
+import os
+import uuid
+import stat
+import errno
+import logging
+from contextlib import contextmanager
+
+import cephfs
+
+from .index import Index
+from ..exception import IndexException, VolumeException
+from ..fs_util import list_one_entry_at_a_time
+
+log = logging.getLogger(__name__)
+
+class CloneIndex(Index):
+    SUB_GROUP_NAME = "clone"
+    PATH_MAX = 4096
+
+    @property
+    def path(self):
+        return os.path.join(super(CloneIndex, self).path, CloneIndex.SUB_GROUP_NAME.encode('utf-8'))
+
+    def _track(self, sink_path):
+        tracking_id = str(uuid.uuid4())
+        source_path = os.path.join(self.path, tracking_id.encode('utf-8'))
+        log.info("tracking-id {0} for path {1}".format(tracking_id, sink_path))
+
+        self.fs.symlink(sink_path, source_path)
+        return tracking_id
+
+    def track(self, sink_path):
+        try:
+            return self._track(sink_path)
+        except (VolumeException, cephfs.Error) as e:
+            if isinstance(e, cephfs.Error):
+                e = IndexException(-e.args[0], e.args[1])
+            elif isinstance(e, VolumeException):
+                e = IndexException(e.errno, e.error_str)
+            raise e
+
+    def untrack(self, tracking_id):
+        log.info("untracking {0}".format(tracking_id))
+        source_path = os.path.join(self.path, tracking_id.encode('utf-8'))
+        try:
+            self.fs.unlink(source_path)
+        except cephfs.Error as e:
+            raise IndexException(-e.args[0], e.args[1])
+
+    def get_oldest_clone_entry(self, exclude=[]):
+        min_ctime_entry = None
+        exclude_tracking_ids = [v[0] for v in exclude]
+        log.debug("excluded tracking ids: {0}".format(exclude_tracking_ids))
+        for entry in list_one_entry_at_a_time(self.fs, self.path):
+            dname = entry.d_name
+            dpath = os.path.join(self.path, dname)
+            st = self.fs.lstat(dpath)
+            if dname not in exclude_tracking_ids and stat.S_ISLNK(st.st_mode):
+                if min_ctime_entry is None or st.st_ctime < min_ctime_entry[1].st_ctime:
+                    min_ctime_entry = (dname, st)
+        if min_ctime_entry:
+            try:
+                linklen = min_ctime_entry[1].st_size
+                sink_path = self.fs.readlink(os.path.join(self.path, min_ctime_entry[0]), CloneIndex.PATH_MAX)
+                return (min_ctime_entry[0], sink_path[:linklen])
+            except cephfs.Error as e:
+                raise IndexException(-e.args[0], e.args[1])
+        return None
+
+    def find_clone_entry_index(self, sink_path):
+        try:
+            for entry in list_one_entry_at_a_time(self.fs, self.path):
+                dname = entry.d_name
+                dpath = os.path.join(self.path, dname)
+                st = self.fs.lstat(dpath)
+                if stat.S_ISLNK(st.st_mode):
+                    target_path = self.fs.readlink(dpath, CloneIndex.PATH_MAX)
+                    if sink_path == target_path[:st.st_size]:
+                        return dname
+            return None
+        except cephfs.Error as e:
+            raise IndexException(-e.args[0], e.args[1])
+
+def create_clone_index(fs, vol_spec):
+    clone_index = CloneIndex(fs, vol_spec)
+    try:
+        fs.mkdirs(clone_index.path, 0o700)
+    except cephfs.Error as e:
+        raise IndexException(-e.args[0], e.args[1])
+
+@contextmanager
+def open_clone_index(fs, vol_spec):
+    clone_index = CloneIndex(fs, vol_spec)
+    try:
+        fs.stat(clone_index.path)
+    except cephfs.Error as e:
+        raise IndexException(-e.args[0], e.args[1])
+    yield clone_index
diff --git a/src/pybind/mgr/volumes/fs/operations/group.py b/src/pybind/mgr/volumes/fs/operations/group.py
new file mode 100644
index 000000000..c91969278
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/group.py
@@ -0,0 +1,301 @@
+import os
+import errno
+import logging
+from contextlib import contextmanager
+
+import cephfs
+
+from .snapshot_util import mksnap, rmsnap
+from .pin_util import pin
+from .template import GroupTemplate
+from ..fs_util import listdir, listsnaps, get_ancestor_xattr, create_base_dir, has_subdir
+from ..exception import VolumeException
+
+log = logging.getLogger(__name__)
+
+class Group(GroupTemplate):
+    # Reserved subvolume group name which we use in paths for subvolumes
+    # that are not assigned to a group (i.e. created with group=None)
+    NO_GROUP_NAME = "_nogroup"
+
+    def __init__(self, fs, vol_spec, groupname):
+        if groupname == Group.NO_GROUP_NAME:
+            raise VolumeException(-errno.EPERM, "Operation not permitted for group '{0}' as it is an internal group.".format(groupname))
+        if groupname in vol_spec.INTERNAL_DIRS:
+            raise VolumeException(-errno.EINVAL, "'{0}' is an internal directory and not a valid group name.".format(groupname))
+        self.fs = fs
+        self.user_id = None
+        self.group_id = None
+        self.vol_spec = vol_spec
+        self.groupname = groupname if groupname else Group.NO_GROUP_NAME
+
+    @property
+    def path(self):
+        return os.path.join(self.vol_spec.base_dir.encode('utf-8'), self.groupname.encode('utf-8'))
+
+    @property
+    def group_name(self):
+        return self.groupname
+
+    @property
+    def uid(self):
+        return self.user_id
+
+    @uid.setter
+    def uid(self, val):
+        self.user_id = val
+
+    @property
+    def gid(self):
+        return self.group_id
+
+    @gid.setter
+    def gid(self, val):
+        self.group_id = val
+
+    def is_default_group(self):
+        return self.groupname == Group.NO_GROUP_NAME
+
+    def list_subvolumes(self):
+        try:
+            return listdir(self.fs, self.path)
+        except VolumeException as ve:
+            # listing a default group when it's not yet created
+            if ve.errno == -errno.ENOENT and self.is_default_group():
+                return []
+            raise
+
+    def has_subvolumes(self):
+        try:
+            return has_subdir(self.fs, self.path)
+        except VolumeException as ve:
+            # listing a default group when it's not yet created
+            if ve.errno == -errno.ENOENT and self.is_default_group():
+                return False
+            raise
+
+    def pin(self, pin_type, pin_setting):
+        return pin(self.fs, self.path, pin_type, pin_setting)
+
+    def create_snapshot(self, snapname):
+        snappath = os.path.join(self.path,
+                                self.vol_spec.snapshot_dir_prefix.encode('utf-8'),
+                                snapname.encode('utf-8'))
+        mksnap(self.fs, snappath)
+
+    def remove_snapshot(self, snapname):
+        snappath = os.path.join(self.path,
+                                self.vol_spec.snapshot_dir_prefix.encode('utf-8'),
+                                snapname.encode('utf-8'))
+        rmsnap(self.fs, snappath)
+
+    def list_snapshots(self):
+        try:
+            dirpath = os.path.join(self.path,
+                                   self.vol_spec.snapshot_dir_prefix.encode('utf-8'))
+            return listsnaps(self.fs, self.vol_spec, dirpath, filter_inherited_snaps=True)
+        except VolumeException as ve:
+            if ve.errno == -errno.ENOENT:
+                return []
+            raise
+
+    def info(self):
+        st = self.fs.statx(self.path, cephfs.CEPH_STATX_BTIME | cephfs.CEPH_STATX_SIZE
+                           | cephfs.CEPH_STATX_UID | cephfs.CEPH_STATX_GID | cephfs.CEPH_STATX_MODE
+                           | cephfs.CEPH_STATX_ATIME | cephfs.CEPH_STATX_MTIME | cephfs.CEPH_STATX_CTIME,
+                           cephfs.AT_SYMLINK_NOFOLLOW)
+        usedbytes = st["size"]
+        try:
+            nsize = int(self.fs.getxattr(self.path, 'ceph.quota.max_bytes').decode('utf-8'))
+        except cephfs.NoData:
+            nsize = 0
+
+        try:
+            data_pool = self.fs.getxattr(self.path, 'ceph.dir.layout.pool').decode('utf-8')
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        return {'uid': int(st["uid"]),
+                'gid': int(st["gid"]),
+                'atime': str(st["atime"]),
+                'mtime': str(st["mtime"]),
+                'ctime': str(st["ctime"]),
+                'mode': int(st["mode"]),
+                'data_pool': data_pool,
+                'created_at': str(st["btime"]),
+                'bytes_quota': "infinite" if nsize == 0 else nsize,
+                'bytes_used': int(usedbytes),
+                'bytes_pcent': "undefined" if nsize == 0 else '{0:.2f}'.format((float(usedbytes) / nsize) * 100.0)}
+
+    def resize(self, newsize, noshrink):
+        try:
+            newsize = int(newsize)
+            if newsize <= 0:
+                raise VolumeException(-errno.EINVAL, "Invalid subvolume group size")
+        except ValueError:
+            newsize = newsize.lower()
+            if not (newsize == "inf" or newsize == "infinite"):
+                raise (VolumeException(-errno.EINVAL, "invalid size option '{0}'".format(newsize)))
+            newsize = 0
+            noshrink = False
+
+        try:
+            maxbytes = int(self.fs.getxattr(self.path, 'ceph.quota.max_bytes').decode('utf-8'))
+        except cephfs.NoData:
+            maxbytes = 0
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        group_stat = self.fs.stat(self.path)
+        if newsize > 0 and newsize < group_stat.st_size:
+            if noshrink:
+                raise VolumeException(-errno.EINVAL, "Can't resize the subvolume group. The new size"
+                                      " '{0}' would be lesser than the current used size '{1}'"
+                                      .format(newsize, group_stat.st_size))
+
+        if not newsize == maxbytes:
+            try:
+                self.fs.setxattr(self.path, 'ceph.quota.max_bytes', str(newsize).encode('utf-8'), 0)
+            except cephfs.Error as e:
+                raise (VolumeException(-e.args[0],
+                                       "Cannot set new size for the subvolume group. '{0}'".format(e.args[1])))
+        return newsize, group_stat.st_size
+
+def set_group_attrs(fs, path, attrs):
+    # set subvolume group attrs
+    # set size
+    quota = attrs.get("quota")
+    if quota is not None:
+        try:
+            fs.setxattr(path, 'ceph.quota.max_bytes', str(quota).encode('utf-8'), 0)
+        except cephfs.InvalidValue:
+            raise VolumeException(-errno.EINVAL, "invalid size specified: '{0}'".format(quota))
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    # set pool layout
+    pool = attrs.get("data_pool")
+    if not pool:
+        pool = get_ancestor_xattr(fs, path, "ceph.dir.layout.pool")
+    try:
+        fs.setxattr(path, 'ceph.dir.layout.pool', pool.encode('utf-8'), 0)
+    except cephfs.InvalidValue:
+        raise VolumeException(-errno.EINVAL,
+                              "Invalid pool layout '{0}'. It must be a valid data pool".format(pool))
+
+    # set uid/gid
+    uid = attrs.get("uid")
+    if uid is None:
+        uid = 0
+    else:
+        try:
+            uid = int(uid)
+            if uid < 0:
+                raise ValueError
+        except ValueError:
+            raise VolumeException(-errno.EINVAL, "invalid UID")
+
+    gid = attrs.get("gid")
+    if gid is None:
+        gid = 0
+    else:
+        try:
+            gid = int(gid)
+            if gid < 0:
+                raise ValueError
+        except ValueError:
+            raise VolumeException(-errno.EINVAL, "invalid GID")
+    fs.chown(path, uid, gid)
+
+    # set mode
+    mode = attrs.get("mode", None)
+    if mode is not None:
+        fs.lchmod(path, mode)
+
+def create_group(fs, vol_spec, groupname, size, pool, mode, uid, gid):
+    """
+    create a subvolume group.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param groupname: subvolume group name
+    :param size: In bytes, or None for no size limit
+    :param pool: the RADOS pool where the data objects of the subvolumes will be stored
+    :param mode: the user permissions
+    :param uid: the user identifier
+    :param gid: the group identifier
+    :return: None
+    """
+    group = Group(fs, vol_spec, groupname)
+    path = group.path
+    vol_spec_base_dir = group.vol_spec.base_dir.encode('utf-8')
+
+    # create vol_spec base directory with default mode(0o755) if it doesn't exist
+    create_base_dir(fs, vol_spec_base_dir, vol_spec.DEFAULT_MODE)
+    fs.mkdir(path, mode)
+    try:
+        attrs = {
+            'uid': uid,
+            'gid': gid,
+            'data_pool': pool,
+            'quota': size
+        }
+        set_group_attrs(fs, path, attrs)
+    except (cephfs.Error, VolumeException) as e:
+        try:
+            # cleanup group path on best effort basis
+            log.debug("cleaning up subvolume group path: {0}".format(path))
+            fs.rmdir(path)
+        except cephfs.Error as ce:
+            log.debug("failed to clean up subvolume group {0} with path: {1} ({2})".format(groupname, path, ce))
+        if isinstance(e, cephfs.Error):
+            e = VolumeException(-e.args[0], e.args[1])
+        raise e
+
+def remove_group(fs, vol_spec, groupname):
+    """
+    remove a subvolume group.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param groupname: subvolume group name
+    :return: None
+    """
+    group = Group(fs, vol_spec, groupname)
+    try:
+        fs.rmdir(group.path)
+    except cephfs.Error as e:
+        if e.args[0] == errno.ENOENT:
+            raise VolumeException(-errno.ENOENT, "subvolume group '{0}' does not exist".format(groupname))
+        raise VolumeException(-e.args[0], e.args[1])
+
+@contextmanager
+def open_group(fs, vol_spec, groupname):
+    """
+    open a subvolume group. This API is to be used as a context manager.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param groupname: subvolume group name
+    :return: yields a group object (subclass of GroupTemplate)
+    """
+    group = Group(fs, vol_spec, groupname)
+    try:
+        st = fs.stat(group.path)
+        group.uid = int(st.st_uid)
+        group.gid = int(st.st_gid)
+    except cephfs.Error as e:
+        if e.args[0] == errno.ENOENT:
+            if not group.is_default_group():
+                raise VolumeException(-errno.ENOENT, "subvolume group '{0}' does not exist".format(groupname))
+        else:
+            raise VolumeException(-e.args[0], e.args[1])
+    yield group
+
+@contextmanager
+def open_group_unique(fs, vol_spec, groupname, c_group, c_groupname):
+    if groupname == c_groupname:
+        yield c_group
+    else:
+        with open_group(fs, vol_spec, groupname) as group:
+            yield group
diff --git a/src/pybind/mgr/volumes/fs/operations/index.py b/src/pybind/mgr/volumes/fs/operations/index.py
new file mode 100644
index 000000000..0e4296d75
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/index.py
@@ -0,0 +1,23 @@
+import errno
+import os
+
+from ..exception import VolumeException
+from .template import GroupTemplate
+
+class Index(GroupTemplate):
+    GROUP_NAME = "_index"
+
+    def __init__(self, fs, vol_spec):
+        self.fs = fs
+        self.vol_spec = vol_spec
+        self.groupname = Index.GROUP_NAME
+
+    @property
+    def path(self):
+        return os.path.join(self.vol_spec.base_dir.encode('utf-8'), self.groupname.encode('utf-8'))
+
+    def track(self, *args):
+        raise VolumeException(-errno.EINVAL, "operation not supported.")
+
+    def untrack(self, tracking_id):
+        raise VolumeException(-errno.EINVAL, "operation not supported.")
diff --git a/src/pybind/mgr/volumes/fs/operations/lock.py b/src/pybind/mgr/volumes/fs/operations/lock.py
new file mode 100644
index 000000000..7ef6923e1
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/lock.py
@@ -0,0 +1,43 @@
+from contextlib import contextmanager
+import logging
+from threading import Lock
+from typing import Dict
+
+log = logging.getLogger(__name__)
+
+# singleton design pattern taken from http://www.aleax.it/5ep.html
+
+class GlobalLock(object):
+    """
+    Global lock to serialize operations in mgr/volumes. This lock
+    is currently held when accessing (opening) a volume to perform
+    group/subvolume operations. Since this is a big lock, it's rather
+    inefficient -- but right now it's ok since mgr/volumes does not
+    expect concurrent operations via its APIs.
+
+    As and when features get added (such as clone, where mgr/volumes
+    would maintain subvolume states in the filesystem), there might
+    be a need to allow concurrent operations. In that case it would
+    be nice to implement an efficient path based locking mechanism.
+
+    See: https://people.eecs.berkeley.edu/~kubitron/courses/cs262a-F14/projects/reports/project6_report.pdf
+    """
+    _shared_state = {
+        'lock' : Lock(),
+        'init' : False
+    } # type: Dict
+
+    def __init__(self):
+        with self._shared_state['lock']:
+            if not self._shared_state['init']:
+                self._shared_state['init'] = True
+        # share this state among all instances
+        self.__dict__ = self._shared_state
+
+    @contextmanager
+    def lock_op(self):
+        log.debug("entering global lock")
+        with self._shared_state['lock']:
+            log.debug("acquired global lock")
+            yield
+        log.debug("exited global lock")
diff --git a/src/pybind/mgr/volumes/fs/operations/pin_util.py b/src/pybind/mgr/volumes/fs/operations/pin_util.py
new file mode 100644
index 000000000..9ea79e546
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/pin_util.py
@@ -0,0 +1,34 @@
+import os
+import errno
+
+import cephfs
+
+from ..exception import VolumeException
+from distutils.util import strtobool
+
+_pin_value = {
+    "export": lambda x: int(x),
+    "distributed": lambda x: int(strtobool(x)),
+    "random": lambda x: float(x),
+}
+_pin_xattr = {
+    "export": "ceph.dir.pin",
+    "distributed": "ceph.dir.pin.distributed",
+    "random": "ceph.dir.pin.random",
+}
+
+def pin(fs, path, pin_type, pin_setting):
+    """
+    Set a pin on a directory.
+    """
+    assert pin_type in _pin_xattr
+
+    try:
+        pin_setting = _pin_value[pin_type](pin_setting)
+    except ValueError as e:
+        raise VolumeException(-errno.EINVAL, f"pin value wrong type: {pin_setting}")
+
+    try:
+        fs.setxattr(path, _pin_xattr[pin_type], str(pin_setting).encode('utf-8'), 0)
+    except cephfs.Error as e:
+        raise VolumeException(-e.args[0], e.args[1])
diff --git a/src/pybind/mgr/volumes/fs/operations/rankevicter.py b/src/pybind/mgr/volumes/fs/operations/rankevicter.py
new file mode 100644
index 000000000..5b945c389
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/rankevicter.py
@@ -0,0 +1,114 @@
+import errno
+import json
+import logging
+import threading
+import time
+
+from .volume import get_mds_map
+from ..exception import ClusterTimeout, ClusterError
+
+log = logging.getLogger(__name__)
+
+class RankEvicter(threading.Thread):
+    """
+    Thread for evicting client(s) from a particular MDS daemon instance.
+
+    This is more complex than simply sending a command, because we have to
+    handle cases where MDS daemons might not be fully up yet, and/or might
+    be transiently unresponsive to commands.
+    """
+    class GidGone(Exception):
+        pass
+
+    POLL_PERIOD = 5
+
+    def __init__(self, mgr, fs, client_spec, volname, rank, gid, mds_map, ready_timeout):
+        """
+        :param client_spec: list of strings, used as filter arguments to "session evict"
+                            pass ["id=123"] to evict a single client with session id 123.
+        """
+        self.volname = volname
+        self.rank = rank
+        self.gid = gid
+        self._mds_map = mds_map
+        self._client_spec = client_spec
+        self._fs = fs
+        self._ready_timeout = ready_timeout
+        self._ready_waited = 0
+        self.mgr = mgr
+
+        self.success = False
+        self.exception = None
+
+        super(RankEvicter, self).__init__()
+
+    def _ready_to_evict(self):
+        if self._mds_map['up'].get("mds_{0}".format(self.rank), None) != self.gid:
+            log.info("Evicting {0} from {1}/{2}: rank no longer associated with gid, done.".format(
+                self._client_spec, self.rank, self.gid
+            ))
+            raise RankEvicter.GidGone()
+
+        info = self._mds_map['info']["gid_{0}".format(self.gid)]
+        log.debug("_ready_to_evict: state={0}".format(info['state']))
+        return info['state'] in ["up:active", "up:clientreplay"]
+
+    def _wait_for_ready(self):
+        """
+        Wait for that MDS rank to reach an active or clientreplay state, and
+        not be laggy.
+        """
+        while not self._ready_to_evict():
+            if self._ready_waited > self._ready_timeout:
+                raise ClusterTimeout()
+
+            time.sleep(self.POLL_PERIOD)
+            self._ready_waited += self.POLL_PERIOD
+            self._mds_map = get_mds_map(self.mgr, self.volname)
+
+    def _evict(self):
+        """
+        Run the eviction procedure.  Return true on success, false on errors.
+        """
+
+        # Wait til the MDS is believed by the mon to be available for commands
+        try:
+            self._wait_for_ready()
+        except self.GidGone:
+            return True
+
+        # Then send it an evict
+        ret = -errno.ETIMEDOUT
+        while ret == -errno.ETIMEDOUT:
+            log.debug("mds_command: {0}, {1}".format(
+                "%s" % self.gid, ["session", "evict"] + self._client_spec
+            ))
+            ret, outb, outs = self._fs.mds_command(
+                "%s" % self.gid,
+                json.dumps({
+                                "prefix": "session evict",
+                                "filters": self._client_spec
+                }), "")
+            log.debug("mds_command: complete {0} {1}".format(ret, outs))
+
+            # If we get a clean response, great, it's gone from that rank.
+            if ret == 0:
+                return True
+            elif ret == -errno.ETIMEDOUT:
+                # Oh no, the MDS went laggy (that's how libcephfs knows to emit this error)
+                self._mds_map = get_mds_map(self.mgr, self.volname)
+                try:
+                    self._wait_for_ready()
+                except self.GidGone:
+                    return True
+            else:
+                raise ClusterError("Sending evict to mds.{0}".format(self.gid), ret, outs)
+
+    def run(self):
+        try:
+            self._evict()
+        except Exception as e:
+            self.success = False
+            self.exception = e
+        else:
+            self.success = True
diff --git a/src/pybind/mgr/volumes/fs/operations/resolver.py b/src/pybind/mgr/volumes/fs/operations/resolver.py
new file mode 100644
index 000000000..a9543654e
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/resolver.py
@@ -0,0 +1,26 @@
+import os
+
+from .group import Group
+
+def splitall(path):
+    if path == "/":
+        return ["/"]
+    s = os.path.split(path)
+    return splitall(s[0]) + [s[1]]
+
+def resolve(vol_spec, path):
+    parts = splitall(path)
+    if len(parts) != 4 or os.path.join(parts[0], parts[1]) != vol_spec.subvolume_prefix:
+        return None
+    groupname = None if parts[2] == Group.NO_GROUP_NAME else parts[2]
+    subvolname = parts[3]
+    return (groupname, subvolname)
+
+def resolve_trash(vol_spec, path):
+    parts = splitall(path)
+    if len(parts) != 6 or os.path.join(parts[0], parts[1]) != vol_spec.subvolume_prefix or \
+       parts[4] != '.trash':
+        return None
+    groupname = None if parts[2] == Group.NO_GROUP_NAME else parts[2]
+    subvolname = parts[3]
+    return (groupname, subvolname)
diff --git a/src/pybind/mgr/volumes/fs/operations/snapshot_util.py b/src/pybind/mgr/volumes/fs/operations/snapshot_util.py
new file mode 100644
index 000000000..2223c58e5
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/snapshot_util.py
@@ -0,0 +1,30 @@
+import os
+import errno
+
+import cephfs
+
+from ..exception import VolumeException
+
+def mksnap(fs, snappath):
+    """
+    Create a snapshot, or do nothing if it already exists.
+    """
+    try:
+        # snap create does not accept mode -- use default
+        fs.mkdir(snappath, 0o755)
+    except cephfs.ObjectExists:
+        return
+    except cephfs.Error as e:
+        raise VolumeException(-e.args[0], e.args[1])
+
+def rmsnap(fs, snappath):
+    """
+    Remove a snapshot
+    """
+    try:
+        fs.stat(snappath)
+        fs.rmdir(snappath)
+    except cephfs.ObjectNotFound:
+        raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(os.path.basename(snappath)))
+    except cephfs.Error as e:
+        raise VolumeException(-e.args[0], e.args[1])
diff --git a/src/pybind/mgr/volumes/fs/operations/subvolume.py b/src/pybind/mgr/volumes/fs/operations/subvolume.py
new file mode 100644
index 000000000..dc36477b5
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/subvolume.py
@@ -0,0 +1,74 @@
+import os
+import errno
+from contextlib import contextmanager
+
+from ..exception import VolumeException
+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):
+    """
+    create a subvolume (create a subvolume with the max known version).
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param group: group object for the subvolume
+    :param size: In bytes, or None for no size limit
+    :param isolate_nspace: If true, use separate RADOS namespace for this subvolume
+    :param pool: the RADOS pool where the data objects of the subvolumes will be stored
+    :param mode: the user permissions
+    :param uid: the user identifier
+    :param gid: the group identifier
+    :return: None
+    """
+    subvolume = loaded_subvolumes.get_subvolume_object_max(mgr, fs, vol_spec, group, subvolname)
+    subvolume.create(size, isolate_nspace, pool, mode, uid, gid)
+
+def create_clone(mgr, fs, vol_spec, group, subvolname, pool, source_volume, source_subvolume, snapname):
+    """
+    create a cloned subvolume.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param group: group object for the clone
+    :param subvolname: clone subvolume nam
+    :param pool: the RADOS pool where the data objects of the cloned subvolume will be stored
+    :param source_volume: source subvolumes volume name
+    :param source_subvolume: source (parent) subvolume object
+    :param snapname: source subvolume snapshot
+    :return None
+    """
+    subvolume = loaded_subvolumes.get_subvolume_object_max(mgr, fs, vol_spec, group, subvolname)
+    subvolume.create_clone(pool, source_volume, source_subvolume, snapname)
+
+def remove_subvol(mgr, fs, vol_spec, group, subvolname, force=False, retainsnaps=False):
+    """
+    remove a subvolume.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param group: group object for the subvolume
+    :param subvolname: subvolume name
+    :param force: force remove subvolumes
+    :return: None
+    """
+    op_type = SubvolumeOpType.REMOVE if not force else SubvolumeOpType.REMOVE_FORCE
+    with open_subvol(mgr, fs, vol_spec, group, subvolname, op_type) as subvolume:
+        subvolume.remove(retainsnaps)
+
+@contextmanager
+def open_subvol(mgr, fs, vol_spec, group, subvolname, op_type):
+    """
+    open a subvolume. This API is to be used as a context manager.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :param group: group object for the subvolume
+    :param subvolname: subvolume name
+    :param op_type: operation type for which subvolume is being opened
+    :return: yields a subvolume object (subclass of SubvolumeTemplate)
+    """
+    subvolume = loaded_subvolumes.get_subvolume_object(mgr, fs, vol_spec, group, subvolname)
+    subvolume.open(op_type)
+    yield subvolume
diff --git a/src/pybind/mgr/volumes/fs/operations/template.py b/src/pybind/mgr/volumes/fs/operations/template.py
new file mode 100644
index 000000000..eb55bd743
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/template.py
@@ -0,0 +1,191 @@
+import errno
+
+from enum import Enum, unique
+
+from ..exception import VolumeException
+
+class GroupTemplate(object):
+    def list_subvolumes(self):
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def create_snapshot(self, snapname):
+        """
+        create a subvolume group snapshot.
+
+        :param: group snapshot name
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def remove_snapshot(self, snapname):
+        """
+        remove a subvolume group snapshot.
+
+        :param: group snapshot name
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def list_snapshots(self):
+        """
+        list all subvolume group snapshots.
+
+        :param: None
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+@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'
+    USER_METADATA_SET     = 'user-metadata-set'
+    USER_METADATA_GET     = 'user-metadata-get'
+    USER_METADATA_LIST    = 'user-metadata-ls'
+    USER_METADATA_REMOVE  = 'user-metadata-rm'
+    SNAP_METADATA_SET     = 'snap-metadata-set'
+    SNAP_METADATA_GET     = 'snap-metadata-get'
+    SNAP_METADATA_LIST    = 'snap-metadata-ls'
+    SNAP_METADATA_REMOVE  = 'snap-metadata-rm'
+
+class SubvolumeTemplate(object):
+    VERSION = None # type: int
+
+    @staticmethod
+    def version():
+        return SubvolumeTemplate.VERSION
+
+    def open(self, op_type):
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def status(self):
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def create(self, size, isolate_nspace, pool, mode, uid, gid):
+        """
+        set up metadata, pools and auth for a subvolume.
+
+        This function is idempotent.  It is safe to call this again
+        for an already-created subvolume, even if it is in use.
+
+        :param size: In bytes, or None for no size limit
+        :param isolate_nspace: If true, use separate RADOS namespace for this subvolume
+        :param pool: the RADOS pool where the data objects of the subvolumes will be stored
+        :param mode: the user permissions
+        :param uid: the user identifier
+        :param gid: the group identifier
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def create_clone(self, pool, source_volname, source_subvolume, snapname):
+        """
+        prepare a subvolume to be cloned.
+
+        :param pool: the RADOS pool where the data objects of the cloned subvolume will be stored
+        :param source_volname: source volume of snapshot
+        :param source_subvolume: source subvolume of snapshot
+        :param snapname: snapshot name to be cloned from
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def remove(self):
+        """
+        make a subvolume inaccessible to guests.
+
+        This function is idempotent.  It is safe to call this again
+
+        :param: None
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def resize(self, newsize, nshrink):
+        """
+        resize a subvolume
+
+        :param newsize: new size In bytes (or inf/infinite)
+        :return: new quota size and used bytes as a tuple
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def pin(self, pin_type, pin_setting):
+        """
+        pin a subvolume
+
+        :param pin_type: type of pin
+        :param pin_setting: setting for pin
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def create_snapshot(self, snapname):
+        """
+        snapshot a subvolume.
+
+        :param: subvolume snapshot name
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def remove_snapshot(self, snapname):
+        """
+        remove a subvolume snapshot.
+
+        :param: subvolume snapshot name
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def list_snapshots(self):
+        """
+        list all subvolume snapshots.
+
+        :param: None
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def attach_snapshot(self, snapname, tgt_subvolume):
+        """
+        attach a snapshot to a target cloned subvolume. the target subvolume
+        should be an empty subvolume (type "clone") in "pending" state.
+
+        :param: snapname: snapshot to attach to a clone
+        :param: tgt_subvolume: target clone subvolume
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
+
+    def detach_snapshot(self, snapname, tgt_subvolume):
+        """
+        detach a snapshot from a target cloned subvolume. the target subvolume
+        should either be in "failed" or "completed" state.
+
+        :param: snapname: snapshot to detach from a clone
+        :param: tgt_subvolume: target clone subvolume
+        :return: None
+        """
+        raise VolumeException(-errno.ENOTSUP, "operation not supported.")
diff --git a/src/pybind/mgr/volumes/fs/operations/trash.py b/src/pybind/mgr/volumes/fs/operations/trash.py
new file mode 100644
index 000000000..66f1d71cf
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/trash.py
@@ -0,0 +1,145 @@
+import os
+import uuid
+import logging
+from contextlib import contextmanager
+
+import cephfs
+
+from .template import GroupTemplate
+from ..fs_util import listdir
+from ..exception import VolumeException
+
+log = logging.getLogger(__name__)
+
+class Trash(GroupTemplate):
+    GROUP_NAME = "_deleting"
+
+    def __init__(self, fs, vol_spec):
+        self.fs = fs
+        self.vol_spec = vol_spec
+        self.groupname = Trash.GROUP_NAME
+
+    @property
+    def path(self):
+        return os.path.join(self.vol_spec.base_dir.encode('utf-8'), self.groupname.encode('utf-8'))
+
+    @property
+    def unique_trash_path(self):
+        """
+        return a unique trash directory entry path
+        """
+        return os.path.join(self.path, str(uuid.uuid4()).encode('utf-8'))
+
+    def _get_single_dir_entry(self, exclude_list=[]):
+        exclude_list.extend((b".", b".."))
+        try:
+            with self.fs.opendir(self.path) as d:
+                entry = self.fs.readdir(d)
+                while entry:
+                    if entry.d_name not in exclude_list:
+                        return entry.d_name
+                    entry = self.fs.readdir(d)
+            return None
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def get_trash_entry(self, exclude_list):
+        """
+        get a trash entry excluding entries provided.
+
+        :praram exclude_list: entries to exclude
+        :return: trash entry
+        """
+        return self._get_single_dir_entry(exclude_list)
+
+    def purge(self, trashpath, should_cancel):
+        """
+        purge a trash entry.
+
+        :praram trash_entry: the trash entry to purge
+        :praram should_cancel: callback to check if the purge should be aborted
+        :return: None
+        """
+        def rmtree(root_path):
+            log.debug("rmtree {0}".format(root_path))
+            try:
+                with self.fs.opendir(root_path) as dir_handle:
+                    d = self.fs.readdir(dir_handle)
+                    while d and not should_cancel():
+                        if d.d_name not in (b".", b".."):
+                            d_full = os.path.join(root_path, d.d_name)
+                            if d.is_dir():
+                                rmtree(d_full)
+                            else:
+                                self.fs.unlink(d_full)
+                        d = self.fs.readdir(dir_handle)
+            except cephfs.ObjectNotFound:
+                return
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], e.args[1])
+            # remove the directory only if we were not asked to cancel
+            # (else we would fail to remove this anyway)
+            if not should_cancel():
+                self.fs.rmdir(root_path)
+
+        # catch any unlink errors
+        try:
+            rmtree(trashpath)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def dump(self, path):
+        """
+        move an filesystem entity to trash can.
+
+        :praram path: the filesystem path to be moved
+        :return: None
+        """
+        try:
+            self.fs.rename(path, self.unique_trash_path)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def link(self, path, bname):
+        pth = os.path.join(self.path, bname)
+        try:
+            self.fs.symlink(path, pth)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def delink(self, bname):
+        pth = os.path.join(self.path, bname)
+        try:
+            self.fs.unlink(pth)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+def create_trashcan(fs, vol_spec):
+    """
+    create a trash can.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :return: None
+    """
+    trashcan = Trash(fs, vol_spec)
+    try:
+        fs.mkdirs(trashcan.path, 0o700)
+    except cephfs.Error as e:
+        raise VolumeException(-e.args[0], e.args[1])
+
+@contextmanager
+def open_trashcan(fs, vol_spec):
+    """
+    open a trash can. This API is to be used as a context manager.
+
+    :param fs: ceph filesystem handle
+    :param vol_spec: volume specification
+    :return: yields a trash can object (subclass of GroupTemplate)
+    """
+    trashcan = Trash(fs, vol_spec)
+    try:
+        fs.stat(trashcan.path)
+    except cephfs.Error as e:
+        raise VolumeException(-e.args[0], e.args[1])
+    yield trashcan
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/__init__.py b/src/pybind/mgr/volumes/fs/operations/versions/__init__.py
new file mode 100644
index 000000000..544afa165
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/__init__.py
@@ -0,0 +1,112 @@
+import errno
+import logging
+import importlib
+
+import cephfs
+
+from .subvolume_base import SubvolumeBase
+from .subvolume_attrs import SubvolumeTypes
+from .subvolume_v1 import SubvolumeV1
+from .subvolume_v2 import SubvolumeV2
+from .metadata_manager import MetadataManager
+from .op_sm import SubvolumeOpSm
+from ..template import SubvolumeOpType
+from ...exception import MetadataMgrException, OpSmException, VolumeException
+
+log = logging.getLogger(__name__)
+
+class SubvolumeLoader(object):
+    INVALID_VERSION = -1
+
+    SUPPORTED_MODULES = ['subvolume_v1.SubvolumeV1', 'subvolume_v2.SubvolumeV2']
+
+    def __init__(self):
+        self.max_version = SubvolumeLoader.INVALID_VERSION
+        self.versions = {}
+
+    def _load_module(self, mod_cls):
+        mod_name, cls_name = mod_cls.split('.')
+        mod = importlib.import_module('.versions.{0}'.format(mod_name), package='volumes.fs.operations')
+        return getattr(mod, cls_name)
+
+    def _load_supported_versions(self):
+        for mod_cls in SubvolumeLoader.SUPPORTED_MODULES:
+            cls = self._load_module(mod_cls)
+            log.info("loaded v{0} subvolume".format(cls.version()))
+            if self.max_version is not None or cls.version() > self.max_version:
+                self.max_version = cls.version()
+                self.versions[cls.version()] = cls
+        if self.max_version == SubvolumeLoader.INVALID_VERSION:
+            raise VolumeException(-errno.EINVAL, "no subvolume version available")
+        log.info("max subvolume version is v{0}".format(self.max_version))
+
+    def _get_subvolume_version(self, version):
+        try:
+            return self.versions[version]
+        except KeyError:
+            raise VolumeException(-errno.EINVAL, "subvolume class v{0} does not exist".format(version))
+
+    def get_subvolume_object_max(self, mgr, fs, vol_spec, group, subvolname):
+        return self._get_subvolume_version(self.max_version)(mgr, fs, vol_spec, group, subvolname)
+
+    def upgrade_to_v2_subvolume(self, subvolume):
+        # legacy mode subvolumes cannot be upgraded to v2
+        if subvolume.legacy_mode:
+            return
+
+        version = int(subvolume.metadata_mgr.get_global_option('version'))
+        if version >= SubvolumeV2.version():
+            return
+
+        v1_subvolume = self._get_subvolume_version(version)(subvolume.mgr, subvolume.fs, subvolume.vol_spec, subvolume.group, subvolume.subvolname)
+        try:
+            v1_subvolume.open(SubvolumeOpType.SNAP_LIST)
+        except VolumeException as ve:
+            # if volume is not ready for snapshot listing, do not upgrade at present
+            if ve.errno == -errno.EAGAIN:
+                return
+            raise
+
+        # v1 subvolumes with snapshots cannot be upgraded to v2
+        if v1_subvolume.list_snapshots():
+            return
+
+        subvolume.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_VERSION, SubvolumeV2.version())
+        subvolume.metadata_mgr.flush()
+
+    def upgrade_legacy_subvolume(self, fs, subvolume):
+        assert subvolume.legacy_mode
+        try:
+            fs.mkdirs(subvolume.legacy_dir, 0o700)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], "error accessing subvolume")
+        subvolume_type = SubvolumeTypes.TYPE_NORMAL
+        try:
+            initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
+        except OpSmException as oe:
+            raise VolumeException(-errno.EINVAL, "subvolume creation failed: internal error")
+        qpath = subvolume.base_path.decode('utf-8')
+        # legacy is only upgradable to v1
+        subvolume.init_config(SubvolumeV1.version(), subvolume_type, qpath, initial_state)
+
+    def get_subvolume_object(self, mgr, fs, vol_spec, group, subvolname, upgrade=True):
+        subvolume = SubvolumeBase(mgr, fs, vol_spec, group, subvolname)
+        try:
+            subvolume.discover()
+            self.upgrade_to_v2_subvolume(subvolume)
+            version = int(subvolume.metadata_mgr.get_global_option('version'))
+            subvolume_version_object = self._get_subvolume_version(version)(mgr, fs, vol_spec, group, subvolname, legacy=subvolume.legacy_mode)
+            subvolume_version_object.metadata_mgr.refresh()
+            subvolume_version_object.clean_stale_snapshot_metadata()
+            return subvolume_version_object
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT and upgrade:
+                self.upgrade_legacy_subvolume(fs, subvolume)
+                return self.get_subvolume_object(mgr, fs, vol_spec, group, subvolname, upgrade=False)
+            else:
+                # log the actual error and generalize error string returned to user
+                log.error("error accessing subvolume metadata for '{0}' ({1})".format(subvolname, me))
+                raise VolumeException(-errno.EINVAL, "error accessing subvolume metadata")
+
+loaded_subvolumes = SubvolumeLoader()
+loaded_subvolumes._load_supported_versions()
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/auth_metadata.py b/src/pybind/mgr/volumes/fs/operations/versions/auth_metadata.py
new file mode 100644
index 000000000..259dcd0e0
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/auth_metadata.py
@@ -0,0 +1,208 @@
+from contextlib import contextmanager
+import os
+import fcntl
+import json
+import logging
+import struct
+import uuid
+
+import cephfs
+
+from ..group import Group
+
+log = logging.getLogger(__name__)
+
+class AuthMetadataError(Exception):
+    pass
+
+class AuthMetadataManager(object):
+
+    # Current version
+    version = 6
+
+    # Filename extensions for meta files.
+    META_FILE_EXT = ".meta"
+    DEFAULT_VOL_PREFIX = "/volumes"
+
+    def __init__(self, fs):
+        self.fs = fs
+        self._id = struct.unpack(">Q", uuid.uuid1().bytes[0:8])[0]
+        self.volume_prefix = self.DEFAULT_VOL_PREFIX
+
+    def _to_bytes(self, param):
+        '''
+        Helper method that returns byte representation of the given parameter.
+        '''
+        if isinstance(param, str):
+            return param.encode('utf-8')
+        elif param is None:
+            return param
+        else:
+            return str(param).encode('utf-8')
+
+    def _subvolume_metadata_path(self, group_name, subvol_name):
+        return os.path.join(self.volume_prefix, "_{0}:{1}{2}".format(
+            group_name if group_name != Group.NO_GROUP_NAME else "",
+            subvol_name,
+            self.META_FILE_EXT))
+
+    def _check_compat_version(self, compat_version):
+        if self.version < compat_version:
+            msg = ("The current version of AuthMetadataManager, version {0} "
+                   "does not support the required feature. Need version {1} "
+                   "or greater".format(self.version, compat_version)
+                  )
+            log.error(msg)
+            raise AuthMetadataError(msg)
+
+    def _metadata_get(self, path):
+        """
+        Return a deserialized JSON object, or None
+        """
+        fd = self.fs.open(path, "r")
+        # TODO iterate instead of assuming file < 4MB
+        read_bytes = self.fs.read(fd, 0, 4096 * 1024)
+        self.fs.close(fd)
+        if read_bytes:
+            return json.loads(read_bytes.decode())
+        else:
+            return None
+
+    def _metadata_set(self, path, data):
+        serialized = json.dumps(data)
+        fd = self.fs.open(path, "w")
+        try:
+            self.fs.write(fd, self._to_bytes(serialized), 0)
+            self.fs.fsync(fd, 0)
+        finally:
+            self.fs.close(fd)
+
+    def _lock(self, path):
+        @contextmanager
+        def fn():
+            while(1):
+                fd = self.fs.open(path, os.O_CREAT, 0o755)
+                self.fs.flock(fd, fcntl.LOCK_EX, self._id)
+
+                # The locked file will be cleaned up sometime. It could be
+                # unlinked by consumer e.g., an another manila-share service
+                # instance, before lock was applied on it. Perform checks to
+                # ensure that this does not happen.
+                try:
+                    statbuf = self.fs.stat(path)
+                except cephfs.ObjectNotFound:
+                    self.fs.close(fd)
+                    continue
+
+                fstatbuf = self.fs.fstat(fd)
+                if statbuf.st_ino == fstatbuf.st_ino:
+                    break
+
+            try:
+                yield
+            finally:
+                self.fs.flock(fd, fcntl.LOCK_UN, self._id)
+                self.fs.close(fd)
+
+        return fn()
+
+    def _auth_metadata_path(self, auth_id):
+        return os.path.join(self.volume_prefix, "${0}{1}".format(
+            auth_id, self.META_FILE_EXT))
+
+    def auth_lock(self, auth_id):
+        return self._lock(self._auth_metadata_path(auth_id))
+
+    def auth_metadata_get(self, auth_id):
+        """
+        Call me with the metadata locked!
+
+        Check whether a auth metadata structure can be decoded by the current
+        version of AuthMetadataManager.
+
+        Return auth metadata that the current version of AuthMetadataManager
+        can decode.
+        """
+        auth_metadata = self._metadata_get(self._auth_metadata_path(auth_id))
+
+        if auth_metadata:
+            self._check_compat_version(auth_metadata['compat_version'])
+
+        return auth_metadata
+
+    def auth_metadata_set(self, auth_id, data):
+        """
+        Call me with the metadata locked!
+
+        Fsync the auth metadata.
+
+        Add two version attributes to the auth metadata,
+        'compat_version', the minimum AuthMetadataManager version that can
+        decode the metadata, and 'version', the AuthMetadataManager version
+        that encoded the metadata.
+        """
+        data['compat_version'] = 6
+        data['version'] = self.version
+        return self._metadata_set(self._auth_metadata_path(auth_id), data)
+
+    def create_subvolume_metadata_file(self, group_name, subvol_name):
+        """
+        Create a subvolume metadata file, if it does not already exist, to store
+        data about auth ids having access to the subvolume
+        """
+        fd = self.fs.open(self._subvolume_metadata_path(group_name, subvol_name),
+                          os.O_CREAT, 0o755)
+        self.fs.close(fd)
+
+    def delete_subvolume_metadata_file(self, group_name, subvol_name):
+        vol_meta_path = self._subvolume_metadata_path(group_name, subvol_name)
+        try:
+            self.fs.unlink(vol_meta_path)
+        except cephfs.ObjectNotFound:
+            pass
+
+    def subvol_metadata_lock(self, group_name, subvol_name):
+        """
+        Return a ContextManager which locks the authorization metadata for
+        a particular subvolume, and persists a flag to the metadata indicating
+        that it is currently locked, so that we can detect dirty situations
+        during recovery.
+
+        This lock isn't just to make access to the metadata safe: it's also
+        designed to be used over the two-step process of checking the
+        metadata and then responding to an authorization request, to
+        ensure that at the point we respond the metadata hasn't changed
+        in the background.  It's key to how we avoid security holes
+        resulting from races during that problem ,
+        """
+        return self._lock(self._subvolume_metadata_path(group_name, subvol_name))
+
+    def subvol_metadata_get(self, group_name, subvol_name):
+        """
+        Call me with the metadata locked!
+
+        Check whether a subvolume metadata structure can be decoded by the current
+        version of AuthMetadataManager.
+
+        Return a subvolume_metadata structure that the current version of
+        AuthMetadataManager can decode.
+        """
+        subvolume_metadata = self._metadata_get(self._subvolume_metadata_path(group_name, subvol_name))
+
+        if subvolume_metadata:
+            self._check_compat_version(subvolume_metadata['compat_version'])
+
+        return subvolume_metadata
+
+    def subvol_metadata_set(self, group_name, subvol_name, data):
+        """
+        Call me with the metadata locked!
+
+        Add two version attributes to the subvolume metadata,
+        'compat_version', the minimum AuthMetadataManager version that can
+        decode the metadata and 'version', the AuthMetadataManager version
+        that encoded the metadata.
+        """
+        data['compat_version'] = 1
+        data['version'] = self.version
+        return self._metadata_set(self._subvolume_metadata_path(group_name, subvol_name), data)
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/metadata_manager.py b/src/pybind/mgr/volumes/fs/operations/versions/metadata_manager.py
new file mode 100644
index 000000000..718735d91
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/metadata_manager.py
@@ -0,0 +1,200 @@
+import os
+import errno
+import logging
+import sys
+import threading
+import configparser
+import re
+
+import cephfs
+
+from ...exception import MetadataMgrException
+
+log = logging.getLogger(__name__)
+
+# _lock needs to be shared across all instances of MetadataManager.
+# that is why we have a file level instance
+_lock = threading.Lock()
+
+
+def _conf_reader(fs, fd, offset=0, length=4096):
+    while True:
+        buf = fs.read(fd, offset, length)
+        offset += len(buf)
+        if not buf:
+            return
+        yield buf.decode('utf-8')
+
+
+class _ConfigWriter:
+    def __init__(self, fs, fd):
+        self._fs = fs
+        self._fd = fd
+        self._wrote = 0
+
+    def write(self, value):
+        buf = value.encode('utf-8')
+        wrote = self._fs.write(self._fd, buf, -1)
+        self._wrote += wrote
+        return wrote
+
+    def fsync(self):
+        self._fs.fsync(self._fd, 0)
+
+    @property
+    def wrote(self):
+        return self._wrote
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb):
+        self._fs.close(self._fd)
+
+
+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"
+    GLOBAL_META_KEY_STATE   = "state"
+
+    CLONE_FAILURE_SECTION = "CLONE_FAILURE"
+    CLONE_FAILURE_META_KEY_ERRNO = "errno"
+    CLONE_FAILURE_META_KEY_ERROR_MSG = "error_msg"
+
+    def __init__(self, fs, config_path, mode):
+        self.fs = fs
+        self.mode = mode
+        self.config_path = config_path
+        self.config = configparser.ConfigParser()
+
+    def refresh(self):
+        fd = None
+        try:
+            log.debug("opening config {0}".format(self.config_path))
+            with _lock:
+                fd = self.fs.open(self.config_path, os.O_RDONLY)
+                cfg = ''.join(_conf_reader(self.fs, fd))
+            self.config.read_string(cfg, source=self.config_path)
+        except UnicodeDecodeError:
+            raise MetadataMgrException(-errno.EINVAL,
+                    "failed to decode, erroneous metadata config '{0}'".format(self.config_path))
+        except cephfs.ObjectNotFound:
+            raise MetadataMgrException(-errno.ENOENT, "metadata config '{0}' not found".format(self.config_path))
+        except cephfs.Error as e:
+            raise MetadataMgrException(-e.args[0], e.args[1])
+        except configparser.Error:
+            raise MetadataMgrException(-errno.EINVAL, "failed to parse, erroneous metadata config "
+                    "'{0}'".format(self.config_path))
+        finally:
+            if fd is not None:
+                self.fs.close(fd)
+
+    def flush(self):
+        # cull empty sections
+        for section in list(self.config.sections()):
+            if len(self.config.items(section)) == 0:
+                self.config.remove_section(section)
+
+        try:
+            with _lock:
+                tmp_config_path = self.config_path + b'.tmp'
+                fd = self.fs.open(tmp_config_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, self.mode)
+                with _ConfigWriter(self.fs, fd) as cfg_writer:
+                    self.config.write(cfg_writer)
+                    cfg_writer.fsync()
+                self.fs.rename(tmp_config_path, self.config_path)
+            log.info(f"wrote {cfg_writer.wrote} bytes to config {tmp_config_path}")
+            log.info(f"Renamed {tmp_config_path} to config {self.config_path}")
+        except cephfs.Error as e:
+            raise MetadataMgrException(-e.args[0], e.args[1])
+
+    def init(self, version, typ, path, state):
+        # you may init just once before refresh (helps to overwrite conf)
+        if self.config.has_section(MetadataManager.GLOBAL_SECTION):
+            raise MetadataMgrException(-errno.EINVAL, "init called on an existing config")
+
+        self.add_section(MetadataManager.GLOBAL_SECTION)
+        self.update_section_multi(
+            MetadataManager.GLOBAL_SECTION, {MetadataManager.GLOBAL_META_KEY_VERSION : str(version),
+                                             MetadataManager.GLOBAL_META_KEY_TYPE    : str(typ),
+                                             MetadataManager.GLOBAL_META_KEY_PATH    : str(path),
+                                             MetadataManager.GLOBAL_META_KEY_STATE   : str(state)
+            })
+
+    def add_section(self, section):
+        try:
+            self.config.add_section(section)
+        except configparser.DuplicateSectionError:
+            return
+        except:
+            raise MetadataMgrException(-errno.EINVAL, "error adding section to config")
+
+    def remove_option(self, section, key):
+        if not self.config.has_section(section):
+            raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))
+        return self.config.remove_option(section, key)
+
+    def remove_section(self, section):
+        self.config.remove_section(section)
+
+    def update_section(self, section, key, value):
+        if not self.config.has_section(section):
+            raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))
+        self.config.set(section, key, str(value))
+
+    def update_section_multi(self, section, dct):
+        if not self.config.has_section(section):
+            raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))
+        for key,value in dct.items():
+            self.config.set(section, key, str(value))
+
+    def update_global_section(self, key, value):
+        self.update_section(MetadataManager.GLOBAL_SECTION, key, str(value))
+
+    def get_option(self, section, key):
+        if not self.config.has_section(section):
+            raise MetadataMgrException(-errno.ENOENT, "section '{0}' does not exist".format(section))
+        if not self.config.has_option(section, key):
+            raise MetadataMgrException(-errno.ENOENT, "no config '{0}' in section '{1}'".format(key, section))
+        return self.config.get(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 list_all_keys_with_specified_values_from_section(self, section, value):
+        keys = []
+        if self.config.has_section(section):
+            options = self.config.options(section)
+            for option in options:
+                if (value == self.config.get(section, option)) :
+                    keys.append(option)
+        return keys
+
+    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))
+        return item in [v[1] for v in self.config.items(section)]
+
+    def has_snap_metadata_section(self):
+        sections = self.config.sections()
+        r = re.compile('SNAP_METADATA_.*')
+        for section in sections:
+            if r.match(section):
+                return True
+        return False
+
+    def list_snaps_with_metadata(self):
+        sections = self.config.sections()
+        r = re.compile('SNAP_METADATA_.*')
+        return [section[len("SNAP_METADATA_"):] for section in sections if r.match(section)]
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/op_sm.py b/src/pybind/mgr/volumes/fs/operations/versions/op_sm.py
new file mode 100644
index 000000000..1142600cb
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/op_sm.py
@@ -0,0 +1,114 @@
+import errno
+
+from typing import Dict
+
+from ...exception import OpSmException
+from .subvolume_attrs import SubvolumeTypes, SubvolumeStates, SubvolumeActions
+
+class TransitionKey(object):
+    def __init__(self, subvol_type, state, action_type):
+        self.transition_key = [subvol_type, state, action_type]
+
+    def __hash__(self):
+        return hash(tuple(self.transition_key))
+
+    def __eq__(self, other):
+        return self.transition_key == other.transition_key
+
+    def __neq__(self, other):
+        return not(self == other)
+
+class SubvolumeOpSm(object):
+    transition_table = {} # type: Dict
+
+    @staticmethod
+    def is_complete_state(state):
+        if not isinstance(state, SubvolumeStates):
+            raise OpSmException(-errno.EINVAL, "unknown state '{0}'".format(state))
+        return state == SubvolumeStates.STATE_COMPLETE
+
+    @staticmethod
+    def is_failed_state(state):
+        if not isinstance(state, SubvolumeStates):
+            raise OpSmException(-errno.EINVAL, "unknown state '{0}'".format(state))
+        return state == SubvolumeStates.STATE_FAILED or state == SubvolumeStates.STATE_CANCELED
+
+    @staticmethod
+    def is_init_state(stm_type, state):
+        if not isinstance(state, SubvolumeStates):
+            raise OpSmException(-errno.EINVAL, "unknown state '{0}'".format(state))
+        return state == SubvolumeOpSm.get_init_state(stm_type)
+
+    @staticmethod
+    def get_init_state(stm_type):
+        if not isinstance(stm_type, SubvolumeTypes):
+            raise OpSmException(-errno.EINVAL, "unknown state machine '{0}'".format(stm_type))
+        init_state =  SubvolumeOpSm.transition_table[TransitionKey(stm_type,
+                                                     SubvolumeStates.STATE_INIT,
+                                                     SubvolumeActions.ACTION_NONE)]
+        if not init_state:
+            raise OpSmException(-errno.ENOENT, "initial state for state machine '{0}' not found".format(stm_type))
+        return init_state
+
+    @staticmethod
+    def transition(stm_type, current_state, action):
+        if not isinstance(stm_type, SubvolumeTypes):
+            raise OpSmException(-errno.EINVAL, "unknown state machine '{0}'".format(stm_type))
+        if not isinstance(current_state, SubvolumeStates):
+            raise OpSmException(-errno.EINVAL, "unknown state '{0}'".format(current_state))
+        if not isinstance(action, SubvolumeActions):
+            raise OpSmException(-errno.EINVAL, "unknown action '{0}'".format(action))
+
+        transition = SubvolumeOpSm.transition_table[TransitionKey(stm_type, current_state, action)]
+        if not transition:
+            raise OpSmException(-errno.EINVAL, "invalid action '{0}' on current state {1} for state machine '{2}'".format(action, current_state, stm_type))
+
+        return transition
+
+SubvolumeOpSm.transition_table = {
+    # state transitions for state machine type TYPE_NORMAL
+    TransitionKey(SubvolumeTypes.TYPE_NORMAL,
+                  SubvolumeStates.STATE_INIT,
+                  SubvolumeActions.ACTION_NONE) : SubvolumeStates.STATE_COMPLETE,
+
+    TransitionKey(SubvolumeTypes.TYPE_NORMAL,
+                  SubvolumeStates.STATE_COMPLETE,
+                  SubvolumeActions.ACTION_RETAINED) : SubvolumeStates.STATE_RETAINED,
+
+    # state transitions for state machine type TYPE_CLONE
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_INIT,
+                  SubvolumeActions.ACTION_NONE) : SubvolumeStates.STATE_PENDING,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_PENDING,
+                  SubvolumeActions.ACTION_SUCCESS) : SubvolumeStates.STATE_INPROGRESS,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_PENDING,
+                  SubvolumeActions.ACTION_CANCELLED) : SubvolumeStates.STATE_CANCELED,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_INPROGRESS,
+                  SubvolumeActions.ACTION_SUCCESS) : SubvolumeStates.STATE_COMPLETE,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_INPROGRESS,
+                  SubvolumeActions.ACTION_CANCELLED) : SubvolumeStates.STATE_CANCELED,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_INPROGRESS,
+                  SubvolumeActions.ACTION_FAILED) : SubvolumeStates.STATE_FAILED,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_COMPLETE,
+                  SubvolumeActions.ACTION_RETAINED) : SubvolumeStates.STATE_RETAINED,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_CANCELED,
+                  SubvolumeActions.ACTION_RETAINED) : SubvolumeStates.STATE_RETAINED,
+
+    TransitionKey(SubvolumeTypes.TYPE_CLONE,
+                  SubvolumeStates.STATE_FAILED,
+                  SubvolumeActions.ACTION_RETAINED) : SubvolumeStates.STATE_RETAINED,
+}
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/subvolume_attrs.py b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_attrs.py
new file mode 100644
index 000000000..ec7138cbd
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_attrs.py
@@ -0,0 +1,61 @@
+import errno
+from enum import Enum, unique
+
+from ...exception import VolumeException
+
+@unique
+class SubvolumeTypes(Enum):
+    TYPE_NORMAL  = "subvolume"
+    TYPE_CLONE   = "clone"
+
+    @staticmethod
+    def from_value(value):
+        if value == "subvolume":
+            return SubvolumeTypes.TYPE_NORMAL
+        if value == "clone":
+            return SubvolumeTypes.TYPE_CLONE
+
+        raise VolumeException(-errno.EINVAL, "invalid subvolume type '{0}'".format(value))
+
+@unique
+class SubvolumeStates(Enum):
+    STATE_INIT          = 'init'
+    STATE_PENDING       = 'pending'
+    STATE_INPROGRESS    = 'in-progress'
+    STATE_FAILED        = 'failed'
+    STATE_COMPLETE      = 'complete'
+    STATE_CANCELED      = 'canceled'
+    STATE_RETAINED      = 'snapshot-retained'
+
+    @staticmethod
+    def from_value(value):
+        if value == "init":
+            return SubvolumeStates.STATE_INIT
+        if value == "pending":
+            return SubvolumeStates.STATE_PENDING
+        if value == "in-progress":
+            return SubvolumeStates.STATE_INPROGRESS
+        if value == "failed":
+            return SubvolumeStates.STATE_FAILED
+        if value == "complete":
+            return SubvolumeStates.STATE_COMPLETE
+        if value == "canceled":
+            return SubvolumeStates.STATE_CANCELED
+        if value == "snapshot-retained":
+            return SubvolumeStates.STATE_RETAINED
+
+        raise VolumeException(-errno.EINVAL, "invalid state '{0}'".format(value))
+
+@unique
+class SubvolumeActions(Enum):
+    ACTION_NONE         = 0
+    ACTION_SUCCESS      = 1
+    ACTION_FAILED       = 2
+    ACTION_CANCELLED    = 3
+    ACTION_RETAINED     = 4
+
+@unique
+class SubvolumeFeatures(Enum):
+    FEATURE_SNAPSHOT_CLONE          = "snapshot-clone"
+    FEATURE_SNAPSHOT_RETENTION      = "snapshot-retention"
+    FEATURE_SNAPSHOT_AUTOPROTECT    = "snapshot-autoprotect"
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/subvolume_base.py b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_base.py
new file mode 100644
index 000000000..72fc45a42
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_base.py
@@ -0,0 +1,449 @@
+import os
+import stat
+import uuid
+import errno
+import logging
+import hashlib
+from typing import Dict, Union
+from pathlib import Path
+
+import cephfs
+
+from ..pin_util import pin
+from .subvolume_attrs import SubvolumeTypes, SubvolumeStates
+from .metadata_manager import MetadataManager
+from ..trash import create_trashcan, open_trashcan
+from ...fs_util import get_ancestor_xattr
+from ...exception import MetadataMgrException, VolumeException
+from .op_sm import SubvolumeOpSm
+from .auth_metadata import AuthMetadataManager
+from .subvolume_attrs import SubvolumeStates
+
+log = logging.getLogger(__name__)
+
+class SubvolumeBase(object):
+    LEGACY_CONF_DIR = "_legacy"
+
+    def __init__(self, mgr, fs, vol_spec, group, subvolname, legacy=False):
+        self.mgr = mgr
+        self.fs = fs
+        self.auth_mdata_mgr = AuthMetadataManager(fs)
+        self.cmode = None
+        self.user_id = None
+        self.group_id = None
+        self.vol_spec = vol_spec
+        self.group = group
+        self.subvolname = subvolname
+        self.legacy_mode = legacy
+        self.load_config()
+
+    @property
+    def uid(self):
+        return self.user_id
+
+    @uid.setter
+    def uid(self, val):
+        self.user_id = val
+
+    @property
+    def gid(self):
+        return self.group_id
+
+    @gid.setter
+    def gid(self, val):
+        self.group_id = val
+
+    @property
+    def mode(self):
+        return self.cmode
+
+    @mode.setter
+    def mode(self, val):
+        self.cmode = val
+
+    @property
+    def base_path(self):
+        return os.path.join(self.group.path, self.subvolname.encode('utf-8'))
+
+    @property
+    def config_path(self):
+        return os.path.join(self.base_path, b".meta")
+
+    @property
+    def legacy_dir(self):
+        return os.path.join(self.vol_spec.base_dir.encode('utf-8'), SubvolumeBase.LEGACY_CONF_DIR.encode('utf-8'))
+
+    @property
+    def legacy_config_path(self):
+        try:
+            m = hashlib.md5(self.base_path)
+        except ValueError:
+            try:
+                m = hashlib.md5(self.base_path, usedforsecurity=False) # type: ignore
+            except TypeError:
+                raise VolumeException(-errno.EINVAL,
+                                      "require python's hashlib library to support usedforsecurity flag in FIPS enabled systems")
+
+        meta_config = "{0}.meta".format(m.hexdigest())
+        return os.path.join(self.legacy_dir, meta_config.encode('utf-8'))
+
+    @property
+    def namespace(self):
+        return "{0}{1}".format(self.vol_spec.fs_namespace, self.subvolname)
+
+    @property
+    def group_name(self):
+        return self.group.group_name
+
+    @property
+    def subvol_name(self):
+        return self.subvolname
+
+    @property
+    def legacy_mode(self):
+        return self.legacy
+
+    @legacy_mode.setter
+    def legacy_mode(self, mode):
+        self.legacy = mode
+
+    @property
+    def path(self):
+        """ Path to subvolume data directory """
+        raise NotImplementedError
+
+    @property
+    def features(self):
+        """ List of features supported by the subvolume, containing items from SubvolumeFeatures """
+        raise NotImplementedError
+
+    @property
+    def state(self):
+        """ Subvolume state, one of SubvolumeStates """
+        return SubvolumeStates.from_value(self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_STATE))
+
+    @property
+    def subvol_type(self):
+        return SubvolumeTypes.from_value(self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_TYPE))
+
+    @property
+    def purgeable(self):
+        """ Boolean declaring if subvolume can be purged """
+        raise NotImplementedError
+
+    def clean_stale_snapshot_metadata(self):
+        """ Clean up stale snapshot metadata """
+        raise NotImplementedError
+
+    def load_config(self):
+        try:
+            self.fs.stat(self.legacy_config_path)
+            self.legacy_mode = True
+        except cephfs.Error as e:
+            pass
+
+        log.debug("loading config "
+                  "'{0}' [mode: {1}]".format(self.subvolname, "legacy"
+                                             if self.legacy_mode else "new"))
+        if self.legacy_mode:
+            self.metadata_mgr = MetadataManager(self.fs, self.legacy_config_path, 0o640)
+        else:
+            self.metadata_mgr = MetadataManager(self.fs, self.config_path, 0o640)
+
+    def get_attrs(self, pathname):
+        # get subvolume attributes
+        attrs = {} # type: Dict[str, Union[int, str, None]]
+        stx = self.fs.statx(pathname,
+                            cephfs.CEPH_STATX_UID | cephfs.CEPH_STATX_GID | cephfs.CEPH_STATX_MODE,
+                            cephfs.AT_SYMLINK_NOFOLLOW)
+
+        attrs["uid"] = int(stx["uid"])
+        attrs["gid"] = int(stx["gid"])
+        attrs["mode"] = int(int(stx["mode"]) & ~stat.S_IFMT(stx["mode"]))
+
+        try:
+            attrs["data_pool"] = self.fs.getxattr(pathname, 'ceph.dir.layout.pool').decode('utf-8')
+        except cephfs.NoData:
+            attrs["data_pool"] = None
+
+        try:
+            attrs["pool_namespace"] = self.fs.getxattr(pathname, 'ceph.dir.layout.pool_namespace').decode('utf-8')
+        except cephfs.NoData:
+            attrs["pool_namespace"] = None
+
+        try:
+            attrs["quota"] = int(self.fs.getxattr(pathname, 'ceph.quota.max_bytes').decode('utf-8'))
+        except cephfs.NoData:
+            attrs["quota"] = None
+
+        return attrs
+
+    def set_attrs(self, path, attrs):
+        # set subvolume attributes
+        # set size
+        quota = attrs.get("quota")
+        if quota is not None:
+            try:
+                self.fs.setxattr(path, 'ceph.quota.max_bytes', str(quota).encode('utf-8'), 0)
+            except cephfs.InvalidValue as e:
+                raise VolumeException(-errno.EINVAL, "invalid size specified: '{0}'".format(quota))
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], e.args[1])
+
+        # set pool layout
+        data_pool = attrs.get("data_pool")
+        if data_pool is not None:
+            try:
+                self.fs.setxattr(path, 'ceph.dir.layout.pool', data_pool.encode('utf-8'), 0)
+            except cephfs.InvalidValue:
+                raise VolumeException(-errno.EINVAL,
+                                      "invalid pool layout '{0}' -- need a valid data pool".format(data_pool))
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], e.args[1])
+
+        # isolate namespace
+        xattr_key = xattr_val = None
+        pool_namespace = attrs.get("pool_namespace")
+        if pool_namespace is not None:
+            # enforce security isolation, use separate namespace for this subvolume
+            xattr_key = 'ceph.dir.layout.pool_namespace'
+            xattr_val = pool_namespace
+        elif not data_pool:
+            # If subvolume's namespace layout is not set, then the subvolume's pool
+            # layout remains unset and will undesirably change with ancestor's
+            # pool layout changes.
+            xattr_key = 'ceph.dir.layout.pool'
+            xattr_val = None
+            try:
+                self.fs.getxattr(path, 'ceph.dir.layout.pool').decode('utf-8')
+            except cephfs.NoData as e:
+                xattr_val = get_ancestor_xattr(self.fs, os.path.split(path)[0], "ceph.dir.layout.pool")
+        if xattr_key and xattr_val:
+            try:
+                self.fs.setxattr(path, xattr_key, xattr_val.encode('utf-8'), 0)
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], e.args[1])
+
+        # set uid/gid
+        uid = attrs.get("uid")
+        if uid is None:
+            uid = self.group.uid
+        else:
+            try:
+                if uid < 0:
+                    raise ValueError
+            except ValueError:
+                raise VolumeException(-errno.EINVAL, "invalid UID")
+
+        gid = attrs.get("gid")
+        if gid is None:
+            gid = self.group.gid
+        else:
+            try:
+                if gid < 0:
+                    raise ValueError
+            except ValueError:
+                raise VolumeException(-errno.EINVAL, "invalid GID")
+
+        if uid is not None and gid is not None:
+            self.fs.chown(path, uid, gid)
+
+        # set mode
+        mode = attrs.get("mode", None)
+        if mode is not None:
+            self.fs.lchmod(path, mode)
+
+    def _resize(self, path, newsize, noshrink):
+        try:
+            newsize = int(newsize)
+            if newsize <= 0:
+                raise VolumeException(-errno.EINVAL, "Invalid subvolume size")
+        except ValueError:
+            newsize = newsize.lower()
+            if not (newsize == "inf" or newsize == "infinite"):
+                raise VolumeException(-errno.EINVAL, "invalid size option '{0}'".format(newsize))
+            newsize = 0
+            noshrink = False
+
+        try:
+            maxbytes = int(self.fs.getxattr(path, 'ceph.quota.max_bytes').decode('utf-8'))
+        except cephfs.NoData:
+            maxbytes = 0
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        subvolstat = self.fs.stat(path)
+        if newsize > 0 and newsize < subvolstat.st_size:
+            if noshrink:
+                raise VolumeException(-errno.EINVAL, "Can't resize the subvolume. The new size '{0}' would be lesser than the current "
+                                      "used size '{1}'".format(newsize, subvolstat.st_size))
+
+        if not newsize == maxbytes:
+            try:
+                self.fs.setxattr(path, 'ceph.quota.max_bytes', str(newsize).encode('utf-8'), 0)
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], "Cannot set new size for the subvolume. '{0}'".format(e.args[1]))
+        return newsize, subvolstat.st_size
+
+    def pin(self, pin_type, pin_setting):
+        return pin(self.fs, self.base_path, pin_type, pin_setting)
+
+    def init_config(self, version, subvolume_type, subvolume_path, subvolume_state):
+        self.metadata_mgr.init(version, subvolume_type.value, subvolume_path, subvolume_state.value)
+        self.metadata_mgr.flush()
+
+    def discover(self):
+        log.debug("discovering subvolume '{0}' [mode: {1}]".format(self.subvolname, "legacy" if self.legacy_mode else "new"))
+        try:
+            self.fs.stat(self.base_path)
+            self.metadata_mgr.refresh()
+            log.debug("loaded subvolume '{0}'".format(self.subvolname))
+            subvolpath = self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_PATH)
+            # subvolume with retained snapshots has empty path, don't mistake it for
+            # fabricated metadata.
+            if (not self.legacy_mode and self.state != SubvolumeStates.STATE_RETAINED and
+                self.base_path.decode('utf-8') != str(Path(subvolpath).parent)):
+                raise MetadataMgrException(-errno.ENOENT, 'fabricated .meta')
+        except MetadataMgrException as me:
+            if me.errno in (-errno.ENOENT, -errno.EINVAL) and not self.legacy_mode:
+                log.warn("subvolume '{0}', {1}, "
+                          "assuming legacy_mode".format(self.subvolname, me.error_str))
+                self.legacy_mode = True
+                self.load_config()
+                self.discover()
+            else:
+                raise
+        except cephfs.Error as e:
+            if e.args[0] == errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "subvolume '{0}' does not exist".format(self.subvolname))
+            raise VolumeException(-e.args[0], "error accessing subvolume '{0}'".format(self.subvolname))
+
+    def _trash_dir(self, path):
+        create_trashcan(self.fs, self.vol_spec)
+        with open_trashcan(self.fs, self.vol_spec) as trashcan:
+            trashcan.dump(path)
+            log.info("subvolume path '{0}' moved to trashcan".format(path))
+
+    def _link_dir(self, path, bname):
+        create_trashcan(self.fs, self.vol_spec)
+        with open_trashcan(self.fs, self.vol_spec) as trashcan:
+            trashcan.link(path, bname)
+            log.info("subvolume path '{0}' linked in trashcan bname {1}".format(path, bname))
+
+    def trash_base_dir(self):
+        if self.legacy_mode:
+            self.fs.unlink(self.legacy_config_path)
+        self._trash_dir(self.base_path)
+
+    def create_base_dir(self, mode):
+        try:
+            self.fs.mkdirs(self.base_path, mode)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def info (self):
+        subvolpath = self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_PATH)
+        etype = self.subvol_type
+        st = self.fs.statx(subvolpath, cephfs.CEPH_STATX_BTIME | cephfs.CEPH_STATX_SIZE |
+                                       cephfs.CEPH_STATX_UID | cephfs.CEPH_STATX_GID |
+                                       cephfs.CEPH_STATX_MODE | cephfs.CEPH_STATX_ATIME |
+                                       cephfs.CEPH_STATX_MTIME | cephfs.CEPH_STATX_CTIME,
+                                       cephfs.AT_SYMLINK_NOFOLLOW)
+        usedbytes = st["size"]
+        try:
+            nsize = int(self.fs.getxattr(subvolpath, 'ceph.quota.max_bytes').decode('utf-8'))
+        except cephfs.NoData:
+            nsize = 0
+
+        try:
+            data_pool = self.fs.getxattr(subvolpath, 'ceph.dir.layout.pool').decode('utf-8')
+            pool_namespace = self.fs.getxattr(subvolpath, 'ceph.dir.layout.pool_namespace').decode('utf-8')
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        return {'path': subvolpath, 'type': etype.value, 'uid': int(st["uid"]), 'gid': int(st["gid"]),
+            'atime': str(st["atime"]), 'mtime': str(st["mtime"]), 'ctime': str(st["ctime"]),
+            'mode': int(st["mode"]), 'data_pool': data_pool, 'created_at': str(st["btime"]),
+            'bytes_quota': "infinite" if nsize == 0 else nsize, 'bytes_used': int(usedbytes),
+            'bytes_pcent': "undefined" if nsize == 0 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):
+        try:
+            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()
+        except MetadataMgrException as me:
+            log.error(f"Failed to set user metadata key={keyname} value={value} on subvolume={self.subvol_name} "
+                      f"group={self.group_name} reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+            raise VolumeException(-me.args[0], me.args[1])
+
+    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 does not exist")
+            log.error(f"Failed to remove user metadata key={keyname} on subvolume={self.subvol_name} "
+                      f"group={self.group_name} reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+            raise VolumeException(-me.args[0], me.args[1])
+
+    def get_snap_section_name(self, snapname):
+        section = "SNAP_METADATA" + "_" + snapname;
+        return section;
+
+    def set_snapshot_metadata(self, snapname, keyname, value):
+        try:
+            section = self.get_snap_section_name(snapname)
+            self.metadata_mgr.add_section(section)
+            self.metadata_mgr.update_section(section, keyname, str(value))
+            self.metadata_mgr.flush()
+        except MetadataMgrException as me:
+            log.error(f"Failed to set snapshot metadata key={keyname} value={value} on snap={snapname} "
+                      f"subvolume={self.subvol_name} group={self.group_name} "
+                      f"reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+            raise VolumeException(-me.args[0], me.args[1])
+
+    def get_snapshot_metadata(self, snapname, keyname):
+        try:
+            value = self.metadata_mgr.get_option(self.get_snap_section_name(snapname), keyname)
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "key '{0}' does not exist.".format(keyname))
+            log.error(f"Failed to get snapshot metadata key={keyname} on snap={snapname} "
+                      f"subvolume={self.subvol_name} group={self.group_name} "
+                      f"reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+            raise VolumeException(-me.args[0], me.args[1])
+        return value
+
+    def list_snapshot_metadata(self, snapname):
+        return self.metadata_mgr.list_all_options_from_section(self.get_snap_section_name(snapname))
+
+    def remove_snapshot_metadata(self, snapname, keyname):
+        try:
+            ret = self.metadata_mgr.remove_option(self.get_snap_section_name(snapname), 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, "snapshot metadata not does not exist")
+            log.error(f"Failed to remove snapshot metadata key={keyname} on snap={snapname} "
+                      f"subvolume={self.subvol_name} group={self.group_name} "
+                      f"reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+            raise VolumeException(-me.args[0], me.args[1])
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v1.py b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v1.py
new file mode 100644
index 000000000..b5a10dd6c
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v1.py
@@ -0,0 +1,904 @@
+import os
+import sys
+import stat
+import uuid
+import errno
+import logging
+import json
+from datetime import datetime
+from typing import Any, List, Dict
+from pathlib import Path
+
+import cephfs
+
+from .metadata_manager import MetadataManager
+from .subvolume_attrs import SubvolumeTypes, SubvolumeStates, SubvolumeFeatures
+from .op_sm import SubvolumeOpSm
+from .subvolume_base import SubvolumeBase
+from ..template import SubvolumeTemplate
+from ..snapshot_util import mksnap, rmsnap
+from ..access import allow_access, deny_access
+from ...exception import IndexException, OpSmException, VolumeException, MetadataMgrException, EvictionError
+from ...fs_util import listsnaps, is_inherited_snap, create_base_dir
+from ..template import SubvolumeOpType
+from ..group import Group
+from ..rankevicter import RankEvicter
+from ..volume import get_mds_map
+
+from ..clone_index import open_clone_index, create_clone_index
+
+log = logging.getLogger(__name__)
+
+class SubvolumeV1(SubvolumeBase, SubvolumeTemplate):
+    """
+    Version 1 subvolumes creates a subvolume with path as follows,
+        volumes/<group-name>/<subvolume-name>/<uuid>/
+
+    - The directory under which user data resides is <uuid>
+    - Snapshots of the subvolume are taken within the <uuid> directory
+    - A meta file is maintained under the <subvolume-name> directory as a metadata store, typically storing,
+        - global information about the subvolume (version, path, type, state)
+        - snapshots attached to an ongoing clone operation
+        - clone snapshot source if subvolume is a clone of a snapshot
+    - It retains backward compatability with legacy subvolumes by creating the meta file for legacy subvolumes under
+    /volumes/_legacy/ (see legacy_config_path), thus allowing cloning of older legacy volumes that lack the <uuid>
+    component in the path.
+    """
+    VERSION = 1
+
+    @staticmethod
+    def version():
+        return SubvolumeV1.VERSION
+
+    @property
+    def path(self):
+        try:
+            # no need to stat the path -- open() does that
+            return self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_PATH).encode('utf-8')
+        except MetadataMgrException as me:
+            raise VolumeException(-errno.EINVAL, "error fetching subvolume metadata")
+
+    @property
+    def features(self):
+        return [SubvolumeFeatures.FEATURE_SNAPSHOT_CLONE.value, SubvolumeFeatures.FEATURE_SNAPSHOT_AUTOPROTECT.value]
+
+    def mark_subvolume(self):
+        # set subvolume attr, on subvolume root, marking it as a CephFS subvolume
+        # subvolume root is where snapshots would be taken, and hence is the <uuid> dir for v1 subvolumes
+        try:
+            # MDS treats this as a noop for already marked subvolume
+            self.fs.setxattr(self.path, 'ceph.dir.subvolume', b'1', 0)
+        except cephfs.InvalidValue as e:
+            raise VolumeException(-errno.EINVAL, "invalid value specified for ceph.dir.subvolume")
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def snapshot_base_path(self):
+        """ Base path for all snapshots """
+        return os.path.join(self.path, self.vol_spec.snapshot_dir_prefix.encode('utf-8'))
+
+    def snapshot_path(self, snapname):
+        """ Path to a specific snapshot named 'snapname' """
+        return os.path.join(self.snapshot_base_path(), snapname.encode('utf-8'))
+
+    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):
+        subvolume_type = SubvolumeTypes.TYPE_NORMAL
+        try:
+            initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
+        except OpSmException as oe:
+            raise VolumeException(-errno.EINVAL, "subvolume creation failed: internal error")
+
+        subvol_path = os.path.join(self.base_path, str(uuid.uuid4()).encode('utf-8'))
+        try:
+            # create group directory with default mode(0o755) if it doesn't exist.
+            create_base_dir(self.fs, self.group.path, self.vol_spec.DEFAULT_MODE)
+            # create directory and set attributes
+            self.fs.mkdirs(subvol_path, mode)
+            self.mark_subvolume()
+            attrs = {
+                'uid': uid,
+                'gid': gid,
+                'data_pool': pool,
+                'pool_namespace': self.namespace if isolate_nspace else None,
+                'quota': size
+            }
+            self.set_attrs(subvol_path, attrs)
+
+            # persist subvolume metadata
+            qpath = subvol_path.decode('utf-8')
+            self.init_config(SubvolumeV1.VERSION, subvolume_type, qpath, initial_state)
+        except (VolumeException, MetadataMgrException, cephfs.Error) as e:
+            try:
+                log.info("cleaning up subvolume with path: {0}".format(self.subvolname))
+                self.remove()
+            except VolumeException as ve:
+                log.info("failed to cleanup subvolume '{0}' ({1})".format(self.subvolname, ve))
+
+            if isinstance(e, MetadataMgrException):
+                log.error("metadata manager exception: {0}".format(e))
+                e = VolumeException(-errno.EINVAL, f"exception in subvolume metadata: {os.strerror(-e.args[0])}")
+            elif isinstance(e, cephfs.Error):
+                e = VolumeException(-e.args[0], e.args[1])
+            raise e
+
+    def add_clone_source(self, volname, subvolume, snapname, flush=False):
+        self.metadata_mgr.add_section("source")
+        self.metadata_mgr.update_section("source", "volume", volname)
+        if not subvolume.group.is_default_group():
+            self.metadata_mgr.update_section("source", "group", subvolume.group_name)
+        self.metadata_mgr.update_section("source", "subvolume", subvolume.subvol_name)
+        self.metadata_mgr.update_section("source", "snapshot", snapname)
+        if flush:
+            self.metadata_mgr.flush()
+
+    def remove_clone_source(self, flush=False):
+        self.metadata_mgr.remove_section("source")
+        if flush:
+            self.metadata_mgr.flush()
+
+    def add_clone_failure(self, errno, error_msg):
+        try:
+            self.metadata_mgr.add_section(MetadataManager.CLONE_FAILURE_SECTION)
+            self.metadata_mgr.update_section(MetadataManager.CLONE_FAILURE_SECTION,
+                                             MetadataManager.CLONE_FAILURE_META_KEY_ERRNO, errno)
+            self.metadata_mgr.update_section(MetadataManager.CLONE_FAILURE_SECTION,
+                                             MetadataManager.CLONE_FAILURE_META_KEY_ERROR_MSG, error_msg)
+            self.metadata_mgr.flush()
+        except MetadataMgrException as me:
+            log.error(f"Failed to add clone failure status clone={self.subvol_name} group={self.group_name} "
+                      f"reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+
+    def create_clone(self, pool, source_volname, source_subvolume, snapname):
+        subvolume_type = SubvolumeTypes.TYPE_CLONE
+        try:
+            initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
+        except OpSmException as oe:
+            raise VolumeException(-errno.EINVAL, "clone failed: internal error")
+
+        subvol_path = os.path.join(self.base_path, str(uuid.uuid4()).encode('utf-8'))
+        try:
+            # source snapshot attrs are used to create clone subvolume.
+            # attributes of subvolume's content though, are synced during the cloning process.
+            attrs = source_subvolume.get_attrs(source_subvolume.snapshot_data_path(snapname))
+
+            # The source of the clone may have exceeded its quota limit as
+            # CephFS quotas are imprecise. Cloning such a source may fail if
+            # the quota on the destination is set before starting the clone
+            # copy. So always set the quota on destination after cloning is
+            # successful.
+            attrs["quota"] = None
+
+            # override snapshot pool setting, if one is provided for the clone
+            if pool is not None:
+                attrs["data_pool"] = pool
+                attrs["pool_namespace"] = None
+
+            # create directory and set attributes
+            self.fs.mkdirs(subvol_path, attrs.get("mode"))
+            self.mark_subvolume()
+            self.set_attrs(subvol_path, attrs)
+
+            # persist subvolume metadata and clone source
+            qpath = subvol_path.decode('utf-8')
+            self.metadata_mgr.init(SubvolumeV1.VERSION, subvolume_type.value, qpath, initial_state.value)
+            self.add_clone_source(source_volname, source_subvolume, snapname)
+            self.metadata_mgr.flush()
+        except (VolumeException, MetadataMgrException, cephfs.Error) as e:
+            try:
+                log.info("cleaning up subvolume with path: {0}".format(self.subvolname))
+                self.remove()
+            except VolumeException as ve:
+                log.info("failed to cleanup subvolume '{0}' ({1})".format(self.subvolname, ve))
+
+            if isinstance(e, MetadataMgrException):
+                log.error("metadata manager exception: {0}".format(e))
+                e = VolumeException(-errno.EINVAL, f"exception in subvolume metadata: {os.strerror(-e.args[0])}")
+            elif isinstance(e, cephfs.Error):
+                e = VolumeException(-e.args[0], e.args[1])
+            raise e
+
+    def allowed_ops_by_type(self, vol_type):
+        if vol_type == SubvolumeTypes.TYPE_CLONE:
+            return {op_type for op_type in SubvolumeOpType}
+
+        if vol_type == SubvolumeTypes.TYPE_NORMAL:
+            return {op_type for op_type in SubvolumeOpType} - {SubvolumeOpType.CLONE_STATUS,
+                                                               SubvolumeOpType.CLONE_CANCEL,
+                                                               SubvolumeOpType.CLONE_INTERNAL}
+
+        return {}
+
+    def allowed_ops_by_state(self, vol_state):
+        if vol_state == SubvolumeStates.STATE_COMPLETE:
+            return {op_type for op_type in SubvolumeOpType}
+
+        return {SubvolumeOpType.REMOVE_FORCE,
+                SubvolumeOpType.CLONE_CREATE,
+                SubvolumeOpType.CLONE_STATUS,
+                SubvolumeOpType.CLONE_CANCEL,
+                SubvolumeOpType.CLONE_INTERNAL}
+
+    def open(self, op_type):
+        if not isinstance(op_type, SubvolumeOpType):
+            raise VolumeException(-errno.ENOTSUP, "operation {0} not supported on subvolume '{1}'".format(
+                                  op_type.value, self.subvolname))
+        try:
+            self.metadata_mgr.refresh()
+
+            etype = self.subvol_type
+            if op_type not in self.allowed_ops_by_type(etype):
+                raise VolumeException(-errno.ENOTSUP, "operation '{0}' is not allowed on subvolume '{1}' of type {2}".format(
+                                      op_type.value, self.subvolname, etype.value))
+
+            estate = self.state
+            if op_type not in self.allowed_ops_by_state(estate):
+                raise VolumeException(-errno.EAGAIN, "subvolume '{0}' is not ready for operation {1}".format(
+                                      self.subvolname, op_type.value))
+
+            subvol_path = self.path
+            log.debug("refreshed metadata, checking subvolume path '{0}'".format(subvol_path))
+            st = self.fs.stat(subvol_path)
+            # unconditionally mark as subvolume, to handle pre-existing subvolumes without the mark
+            self.mark_subvolume()
+
+            self.uid = int(st.st_uid)
+            self.gid = int(st.st_gid)
+            self.mode = int(st.st_mode & ~stat.S_IFMT(st.st_mode))
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "subvolume '{0}' does not exist".format(self.subvolname))
+            raise VolumeException(me.args[0], me.args[1])
+        except cephfs.ObjectNotFound:
+            log.debug("missing subvolume path '{0}' for subvolume '{1}'".format(subvol_path, self.subvolname))
+            raise VolumeException(-errno.ENOENT, "mount path missing for subvolume '{0}'".format(self.subvolname))
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def _recover_auth_meta(self, auth_id, auth_meta):
+        """
+        Call me after locking the auth meta file.
+        """
+        remove_subvolumes = []
+
+        for subvol, subvol_data in auth_meta['subvolumes'].items():
+            if not subvol_data['dirty']:
+                continue
+
+            (group_name, subvol_name) = subvol.split('/')
+            group_name = group_name if group_name != 'None' else Group.NO_GROUP_NAME
+            access_level = subvol_data['access_level']
+
+            with self.auth_mdata_mgr.subvol_metadata_lock(group_name, subvol_name):
+                subvol_meta = self.auth_mdata_mgr.subvol_metadata_get(group_name, subvol_name)
+
+                # No SVMeta update indicates that there was no auth update
+                # in Ceph either. So it's safe to remove corresponding
+                # partial update in AMeta.
+                if not subvol_meta or auth_id not in subvol_meta['auths']:
+                    remove_subvolumes.append(subvol)
+                    continue
+
+                want_auth = {
+                    'access_level': access_level,
+                    'dirty': False,
+                }
+                # SVMeta update looks clean. Ceph auth update must have been
+                # clean. Update the dirty flag and continue
+                if subvol_meta['auths'][auth_id] == want_auth:
+                    auth_meta['subvolumes'][subvol]['dirty'] = False
+                    self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+                    continue
+
+                client_entity = "client.{0}".format(auth_id)
+                ret, out, err = self.mgr.mon_command(
+                    {
+                        'prefix': 'auth get',
+                        'entity': client_entity,
+                        'format': 'json'
+                    })
+                if ret == 0:
+                    existing_caps = json.loads(out)
+                elif ret == -errno.ENOENT:
+                    existing_caps = None
+                else:
+                    log.error(err)
+                    raise VolumeException(ret, err)
+
+                self._authorize_subvolume(auth_id, access_level, existing_caps)
+
+            # Recovered from partial auth updates for the auth ID's access
+            # to a subvolume.
+            auth_meta['subvolumes'][subvol]['dirty'] = False
+            self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+        for subvol in remove_subvolumes:
+            del auth_meta['subvolumes'][subvol]
+
+        if not auth_meta['subvolumes']:
+            # Clean up auth meta file
+            self.fs.unlink(self.auth_mdata_mgr._auth_metadata_path(auth_id))
+            return
+
+        # Recovered from all partial auth updates for the auth ID.
+        auth_meta['dirty'] = False
+        self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+    def authorize(self, auth_id, access_level, tenant_id=None, allow_existing_id=False):
+        """
+        Get-or-create a Ceph auth identity for `auth_id` and grant them access
+        to
+        :param auth_id:
+        :param access_level:
+        :param tenant_id: Optionally provide a stringizable object to
+                          restrict any created cephx IDs to other callers
+                          passing the same tenant ID.
+        :allow_existing_id: Optionally authorize existing auth-ids not
+                          created by ceph_volume_client.
+        :return:
+        """
+
+        with self.auth_mdata_mgr.auth_lock(auth_id):
+            client_entity = "client.{0}".format(auth_id)
+            ret, out, err = self.mgr.mon_command(
+                {
+                    'prefix': 'auth get',
+                    'entity': client_entity,
+                    'format': 'json'
+                })
+
+            if ret == 0:
+                existing_caps = json.loads(out)
+            elif ret == -errno.ENOENT:
+                existing_caps = None
+            else:
+                log.error(err)
+                raise VolumeException(ret, err)
+
+            # Existing meta, or None, to be updated
+            auth_meta = self.auth_mdata_mgr.auth_metadata_get(auth_id)
+
+            # subvolume data to be inserted
+            group_name = self.group.groupname if self.group.groupname != Group.NO_GROUP_NAME else None
+            group_subvol_id = "{0}/{1}".format(group_name, self.subvolname)
+            subvolume = {
+                group_subvol_id : {
+                    # The access level at which the auth_id is authorized to
+                    # access the volume.
+                    'access_level': access_level,
+                    'dirty': True,
+                }
+            }
+
+            if auth_meta is None:
+                if not allow_existing_id and existing_caps is not None:
+                    msg = "auth ID: {0} exists and not created by mgr plugin. Not allowed to modify".format(auth_id)
+                    log.error(msg)
+                    raise VolumeException(-errno.EPERM, msg)
+
+                # non-existent auth IDs
+                sys.stderr.write("Creating meta for ID {0} with tenant {1}\n".format(
+                    auth_id, tenant_id
+                ))
+                log.debug("Authorize: no existing meta")
+                auth_meta = {
+                    'dirty': True,
+                    'tenant_id': str(tenant_id) if tenant_id else None,
+                    'subvolumes': subvolume
+                }
+            else:
+                # Update 'volumes' key (old style auth metadata file) to 'subvolumes' key
+                if 'volumes' in auth_meta:
+                    auth_meta['subvolumes'] = auth_meta.pop('volumes')
+
+                # Disallow tenants to share auth IDs
+                if str(auth_meta['tenant_id']) != str(tenant_id):
+                    msg = "auth ID: {0} is already in use".format(auth_id)
+                    log.error(msg)
+                    raise VolumeException(-errno.EPERM, msg)
+
+                if auth_meta['dirty']:
+                    self._recover_auth_meta(auth_id, auth_meta)
+
+                log.debug("Authorize: existing tenant {tenant}".format(
+                    tenant=auth_meta['tenant_id']
+                ))
+                auth_meta['dirty'] = True
+                auth_meta['subvolumes'].update(subvolume)
+
+            self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+            with self.auth_mdata_mgr.subvol_metadata_lock(self.group.groupname, self.subvolname):
+                key = self._authorize_subvolume(auth_id, access_level, existing_caps)
+
+            auth_meta['dirty'] = False
+            auth_meta['subvolumes'][group_subvol_id]['dirty'] = False
+            self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+            if tenant_id:
+                return key
+            else:
+                # Caller wasn't multi-tenant aware: be safe and don't give
+                # them a key
+                return ""
+
+    def _authorize_subvolume(self, auth_id, access_level, existing_caps):
+        subvol_meta = self.auth_mdata_mgr.subvol_metadata_get(self.group.groupname, self.subvolname)
+
+        auth = {
+            auth_id: {
+                'access_level': access_level,
+                'dirty': True,
+            }
+        }
+
+        if subvol_meta is None:
+            subvol_meta = {
+                'auths': auth
+            }
+        else:
+            subvol_meta['auths'].update(auth)
+            self.auth_mdata_mgr.subvol_metadata_set(self.group.groupname, self.subvolname, subvol_meta)
+
+        key = self._authorize(auth_id, access_level, existing_caps)
+
+        subvol_meta['auths'][auth_id]['dirty'] = False
+        self.auth_mdata_mgr.subvol_metadata_set(self.group.groupname, self.subvolname, subvol_meta)
+
+        return key
+
+    def _authorize(self, auth_id, access_level, existing_caps):
+        subvol_path = self.path
+        log.debug("Authorizing Ceph id '{0}' for path '{1}'".format(auth_id, subvol_path))
+
+        # First I need to work out what the data pool is for this share:
+        # read the layout
+        try:
+            pool = self.fs.getxattr(subvol_path, 'ceph.dir.layout.pool').decode('utf-8')
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        try:
+            namespace = self.fs.getxattr(subvol_path, 'ceph.dir.layout.pool_namespace').decode('utf-8')
+        except cephfs.NoData:
+            namespace = None
+
+        # Now construct auth capabilities that give the guest just enough
+        # permissions to access the share
+        client_entity = "client.{0}".format(auth_id)
+        want_mds_cap = "allow {0} path={1}".format(access_level, subvol_path.decode('utf-8'))
+        want_osd_cap = "allow {0} pool={1}{2}".format(
+                access_level, pool, " namespace={0}".format(namespace) if namespace else "")
+
+        # Construct auth caps that if present might conflict with the desired
+        # auth caps.
+        unwanted_access_level = 'r' if access_level == 'rw' else 'rw'
+        unwanted_mds_cap = 'allow {0} path={1}'.format(unwanted_access_level, subvol_path.decode('utf-8'))
+        unwanted_osd_cap = "allow {0} pool={1}{2}".format(
+                unwanted_access_level, pool, " namespace={0}".format(namespace) if namespace else "")
+
+        return allow_access(self.mgr, client_entity, want_mds_cap, want_osd_cap,
+                            unwanted_mds_cap, unwanted_osd_cap, existing_caps)
+
+    def deauthorize(self, auth_id):
+        with self.auth_mdata_mgr.auth_lock(auth_id):
+            # Existing meta, or None, to be updated
+            auth_meta = self.auth_mdata_mgr.auth_metadata_get(auth_id)
+
+            if auth_meta is None:
+                msg = "auth ID: {0} doesn't exist".format(auth_id)
+                log.error(msg)
+                raise VolumeException(-errno.ENOENT, msg)
+
+            # Update 'volumes' key (old style auth metadata file) to 'subvolumes' key
+            if 'volumes' in auth_meta:
+                auth_meta['subvolumes'] = auth_meta.pop('volumes')
+
+            group_name = self.group.groupname if self.group.groupname != Group.NO_GROUP_NAME else None
+            group_subvol_id = "{0}/{1}".format(group_name, self.subvolname)
+            if (auth_meta is None) or (not auth_meta['subvolumes']):
+                log.warning("deauthorized called for already-removed auth"
+                         "ID '{auth_id}' for subvolume '{subvolume}'".format(
+                    auth_id=auth_id, subvolume=self.subvolname
+                ))
+                # Clean up the auth meta file of an auth ID
+                self.fs.unlink(self.auth_mdata_mgr._auth_metadata_path(auth_id))
+                return
+
+            if group_subvol_id not in auth_meta['subvolumes']:
+                log.warning("deauthorized called for already-removed auth"
+                         "ID '{auth_id}' for subvolume '{subvolume}'".format(
+                    auth_id=auth_id, subvolume=self.subvolname
+                ))
+                return
+
+            if auth_meta['dirty']:
+                self._recover_auth_meta(auth_id, auth_meta)
+
+            auth_meta['dirty'] = True
+            auth_meta['subvolumes'][group_subvol_id]['dirty'] = True
+            self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+            self._deauthorize_subvolume(auth_id)
+
+            # Filter out the volume we're deauthorizing
+            del auth_meta['subvolumes'][group_subvol_id]
+
+            # Clean up auth meta file
+            if not auth_meta['subvolumes']:
+                self.fs.unlink(self.auth_mdata_mgr._auth_metadata_path(auth_id))
+                return
+
+            auth_meta['dirty'] = False
+            self.auth_mdata_mgr.auth_metadata_set(auth_id, auth_meta)
+
+    def _deauthorize_subvolume(self, auth_id):
+        with self.auth_mdata_mgr.subvol_metadata_lock(self.group.groupname, self.subvolname):
+            subvol_meta = self.auth_mdata_mgr.subvol_metadata_get(self.group.groupname, self.subvolname)
+
+            if (subvol_meta is None) or (auth_id not in subvol_meta['auths']):
+                log.warning("deauthorized called for already-removed auth"
+                         "ID '{auth_id}' for subvolume '{subvolume}'".format(
+                    auth_id=auth_id, subvolume=self.subvolname
+                ))
+                return
+
+            subvol_meta['auths'][auth_id]['dirty'] = True
+            self.auth_mdata_mgr.subvol_metadata_set(self.group.groupname, self.subvolname, subvol_meta)
+
+            self._deauthorize(auth_id)
+
+            # Remove the auth_id from the metadata *after* removing it
+            # from ceph, so that if we crashed here, we would actually
+            # recreate the auth ID during recovery (i.e. end up with
+            # a consistent state).
+
+            # Filter out the auth we're removing
+            del subvol_meta['auths'][auth_id]
+            self.auth_mdata_mgr.subvol_metadata_set(self.group.groupname, self.subvolname, subvol_meta)
+
+    def _deauthorize(self, auth_id):
+        """
+        The volume must still exist.
+        """
+        client_entity = "client.{0}".format(auth_id)
+        subvol_path = self.path
+        try:
+            pool_name = self.fs.getxattr(subvol_path, 'ceph.dir.layout.pool').decode('utf-8')
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+        try:
+            namespace = self.fs.getxattr(subvol_path, 'ceph.dir.layout.pool_namespace').decode('utf-8')
+        except cephfs.NoData:
+            namespace = None
+
+        # The auth_id might have read-only or read-write mount access for the
+        # subvolume path.
+        access_levels = ('r', 'rw')
+        want_mds_caps = ['allow {0} path={1}'.format(access_level, subvol_path.decode('utf-8'))
+                         for access_level in access_levels]
+        want_osd_caps = ['allow {0} pool={1}{2}'.format(
+                          access_level, pool_name, " namespace={0}".format(namespace) if namespace else "")
+                         for access_level in access_levels]
+        deny_access(self.mgr, client_entity, want_mds_caps, want_osd_caps)
+
+    def authorized_list(self):
+        """
+        Expose a list of auth IDs that have access to a subvolume.
+
+        return: a list of (auth_id, access_level) tuples, where
+                the access_level can be 'r' , or 'rw'.
+                None if no auth ID is given access to the subvolume.
+        """
+        with self.auth_mdata_mgr.subvol_metadata_lock(self.group.groupname, self.subvolname):
+            meta = self.auth_mdata_mgr.subvol_metadata_get(self.group.groupname, self.subvolname)
+            auths = [] # type: List[Dict[str,str]]
+            if not meta or not meta['auths']:
+                return auths
+
+            for auth, auth_data in meta['auths'].items():
+                # Skip partial auth updates.
+                if not auth_data['dirty']:
+                    auths.append({auth: auth_data['access_level']})
+
+            return auths
+
+    def evict(self, volname, auth_id, timeout=30):
+        """
+        Evict all clients based on the authorization ID and the subvolume path mounted.
+        Assumes that the authorization key has been revoked prior to calling this function.
+
+        This operation can throw an exception if the mon cluster is unresponsive, or
+        any individual MDS daemon is unresponsive for longer than the timeout passed in.
+        """
+
+        client_spec = ["auth_name={0}".format(auth_id), ]
+        client_spec.append("client_metadata.root={0}".
+                           format(self.path.decode('utf-8')))
+
+        log.info("evict clients with {0}".format(', '.join(client_spec)))
+
+        mds_map = get_mds_map(self.mgr, volname)
+        if not mds_map:
+            raise VolumeException(-errno.ENOENT, "mdsmap for volume {0} not found".format(volname))
+
+        up = {}
+        for name, gid in mds_map['up'].items():
+            # Quirk of the MDSMap JSON dump: keys in the up dict are like "mds_0"
+            assert name.startswith("mds_")
+            up[int(name[4:])] = gid
+
+        # For all MDS ranks held by a daemon
+        # Do the parallelism in python instead of using "tell mds.*", because
+        # the latter doesn't give us per-mds output
+        threads = []
+        for rank, gid in up.items():
+            thread = RankEvicter(self.mgr, self.fs, client_spec, volname, rank, gid, mds_map, timeout)
+            thread.start()
+            threads.append(thread)
+
+        for t in threads:
+            t.join()
+
+        log.info("evict: joined all")
+
+        for t in threads:
+            if not t.success:
+                msg = ("Failed to evict client with {0} from mds {1}/{2}: {3}".
+                       format(', '.join(client_spec), t.rank, t.gid, t.exception)
+                      )
+                log.error(msg)
+                raise EvictionError(msg)
+
+    def _get_clone_source(self):
+        try:
+            clone_source = {
+                'volume'   : self.metadata_mgr.get_option("source", "volume"),
+                'subvolume': self.metadata_mgr.get_option("source", "subvolume"),
+                'snapshot' : self.metadata_mgr.get_option("source", "snapshot"),
+            }
+
+            try:
+                clone_source["group"] = self.metadata_mgr.get_option("source", "group")
+            except MetadataMgrException as me:
+                if me.errno == -errno.ENOENT:
+                    pass
+                else:
+                    raise
+        except MetadataMgrException as me:
+            raise VolumeException(-errno.EINVAL, "error fetching subvolume metadata")
+        return clone_source
+
+    def _get_clone_failure(self):
+        clone_failure = {
+            'errno'     : self.metadata_mgr.get_option(MetadataManager.CLONE_FAILURE_SECTION, MetadataManager.CLONE_FAILURE_META_KEY_ERRNO),
+            'error_msg' : self.metadata_mgr.get_option(MetadataManager.CLONE_FAILURE_SECTION, MetadataManager.CLONE_FAILURE_META_KEY_ERROR_MSG),
+        }
+        return clone_failure
+
+    @property
+    def status(self):
+        state = SubvolumeStates.from_value(self.metadata_mgr.get_global_option(MetadataManager.GLOBAL_META_KEY_STATE))
+        subvolume_type = self.subvol_type
+        subvolume_status = {
+            'state' : state.value
+        }
+        if not SubvolumeOpSm.is_complete_state(state) and subvolume_type == SubvolumeTypes.TYPE_CLONE:
+            subvolume_status["source"] = self._get_clone_source()
+        if SubvolumeOpSm.is_failed_state(state) and subvolume_type == SubvolumeTypes.TYPE_CLONE:
+            try:
+                subvolume_status["failure"] = self._get_clone_failure()
+            except MetadataMgrException:
+                pass
+
+        return subvolume_status
+
+    @property
+    def state(self):
+        return super(SubvolumeV1, self).state
+
+    @state.setter
+    def state(self, val):
+        state = val[0].value
+        flush = val[1]
+        self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_STATE, state)
+        if flush:
+            self.metadata_mgr.flush()
+
+    def remove(self, retainsnaps=False):
+        if retainsnaps:
+            raise VolumeException(-errno.EINVAL, "subvolume '{0}' does not support snapshot retention on delete".format(self.subvolname))
+        if self.list_snapshots():
+            raise VolumeException(-errno.ENOTEMPTY, "subvolume '{0}' has snapshots".format(self.subvolname))
+        self.trash_base_dir()
+
+    def resize(self, newsize, noshrink):
+        subvol_path = self.path
+        return self._resize(subvol_path, newsize, noshrink)
+
+    def create_snapshot(self, snapname):
+        try:
+            group_snapshot_path = os.path.join(self.group.path,
+                                               self.vol_spec.snapshot_dir_prefix.encode('utf-8'),
+                                               snapname.encode('utf-8'))
+            self.fs.stat(group_snapshot_path)
+        except cephfs.Error as e:
+            if e.args[0] == errno.ENOENT:
+                snappath = self.snapshot_path(snapname)
+                mksnap(self.fs, snappath)
+            else:
+                raise VolumeException(-e.args[0], e.args[1])
+        else:
+            raise VolumeException(-errno.EINVAL, "subvolumegroup and subvolume snapshot name can't be same")
+
+    def has_pending_clones(self, snapname):
+        try:
+            return self.metadata_mgr.section_has_item('clone snaps', snapname)
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                return False
+            raise
+
+    def get_pending_clones(self, snapname):
+        pending_clones_info = {"has_pending_clones": "no"}  # type: Dict[str, Any]
+        pending_track_id_list = []
+        pending_clone_list = []
+        index_path = ""
+        orphan_clones_count = 0
+
+        try:
+            if self.has_pending_clones(snapname):
+                pending_track_id_list = self.metadata_mgr.list_all_keys_with_specified_values_from_section('clone snaps', snapname)
+            else:
+                return pending_clones_info
+        except MetadataMgrException as me:
+            if me.errno != -errno.ENOENT:
+                raise VolumeException(-me.args[0], me.args[1])
+
+        try:
+            with open_clone_index(self.fs, self.vol_spec) as index:
+                index_path = index.path.decode('utf-8')
+        except IndexException as e:
+            log.warning("failed to open clone index '{0}' for snapshot '{1}'".format(e, snapname))
+            raise VolumeException(-errno.EINVAL, "failed to open clone index")
+
+        for track_id in pending_track_id_list:
+            try:
+                link_path = self.fs.readlink(os.path.join(index_path, track_id), 4096)
+            except cephfs.Error as e:
+                if e.errno != errno.ENOENT:
+                    raise VolumeException(-e.args[0], e.args[1])
+                else:
+                    try:
+                        # If clone is completed between 'list_all_keys_with_specified_values_from_section'
+                        # and readlink(track_id_path) call then readlink will fail with error ENOENT (2)
+                        # Hence we double check whether track_id is exist in .meta file or not.
+                        value = self.metadata_mgr.get_option('clone snaps', track_id)
+                        # Edge case scenario.
+                        # If track_id for clone exist but path /volumes/_index/clone/{track_id} not found
+                        # then clone is orphan.
+                        orphan_clones_count += 1
+                        continue
+                    except MetadataMgrException as me:
+                        if me.errno != -errno.ENOENT:
+                            raise VolumeException(-me.args[0], me.args[1])
+
+            path = Path(link_path.decode('utf-8'))
+            clone_name = os.path.basename(link_path).decode('utf-8')
+            group_name = os.path.basename(path.parent.absolute())
+            details = {"name": clone_name}  # type: Dict[str, str]
+            if group_name != Group.NO_GROUP_NAME:
+                details["target_group"] = group_name
+            pending_clone_list.append(details)
+
+        if len(pending_clone_list) != 0:
+            pending_clones_info["has_pending_clones"] = "yes"
+            pending_clones_info["pending_clones"] = pending_clone_list
+        else:
+            pending_clones_info["has_pending_clones"] = "no"
+
+        if orphan_clones_count > 0:
+            pending_clones_info["orphan_clones_count"] = orphan_clones_count
+
+        return pending_clones_info
+
+    def remove_snapshot(self, snapname, force=False):
+        if self.has_pending_clones(snapname):
+            raise VolumeException(-errno.EAGAIN, "snapshot '{0}' has pending clones".format(snapname))
+        snappath = self.snapshot_path(snapname)
+        try:
+            self.metadata_mgr.remove_section(self.get_snap_section_name(snapname))
+            self.metadata_mgr.flush()
+        except MetadataMgrException as me:
+            if force:
+                log.info(f"Allowing snapshot removal on failure of it's metadata removal with force on "
+                         f"snap={snapname} subvol={self.subvol_name} group={self.group_name} reason={me.args[1]}, "
+                         f"errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+                pass
+            else:
+                log.error(f"Failed to remove snapshot metadata on snap={snapname} subvol={self.subvol_name} "
+                          f"group={self.group_name} reason={me.args[1]}, errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+                raise VolumeException(-errno.EAGAIN,
+                                      f"failed to remove snapshot metadata on snap={snapname} reason={me.args[0]} {me.args[1]}")
+        rmsnap(self.fs, snappath)
+
+    def snapshot_info(self, snapname):
+        if is_inherited_snap(snapname):
+            raise VolumeException(-errno.EINVAL,
+                                  "snapshot name '{0}' is invalid".format(snapname))
+        snappath = self.snapshot_data_path(snapname)
+        snap_info = {}
+        try:
+            snap_attrs = {'created_at':'ceph.snap.btime',
+                          'data_pool':'ceph.dir.layout.pool'}
+            for key, val in snap_attrs.items():
+                snap_info[key] = self.fs.getxattr(snappath, val)
+            pending_clones_info = self.get_pending_clones(snapname)
+            info_dict = {'created_at': str(datetime.fromtimestamp(float(snap_info['created_at']))),
+                    'data_pool': snap_info['data_pool'].decode('utf-8')}  # type: Dict[str, Any]
+            info_dict.update(pending_clones_info);
+            return info_dict
+        except cephfs.Error as e:
+            if e.errno == errno.ENOENT:
+                raise VolumeException(-errno.ENOENT,
+                                      "snapshot '{0}' does not exist".format(snapname))
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def list_snapshots(self):
+        try:
+            dirpath = self.snapshot_base_path()
+            return listsnaps(self.fs, self.vol_spec, dirpath, filter_inherited_snaps=True)
+        except VolumeException as ve:
+            if ve.errno == -errno.ENOENT:
+                return []
+            raise
+
+    def clean_stale_snapshot_metadata(self):
+        """ Clean up stale snapshot metadata """
+        if self.metadata_mgr.has_snap_metadata_section():
+            snap_list = self.list_snapshots()
+            snaps_with_metadata_list = self.metadata_mgr.list_snaps_with_metadata()
+            for snap_with_metadata in snaps_with_metadata_list:
+                if snap_with_metadata.encode('utf-8') not in snap_list:
+                    try:
+                        self.metadata_mgr.remove_section(self.get_snap_section_name(snap_with_metadata))
+                        self.metadata_mgr.flush()
+                    except MetadataMgrException as me:
+                        log.error(f"Failed to remove stale snap metadata on snap={snap_with_metadata} "
+                                  f"subvol={self.subvol_name} group={self.group_name} reason={me.args[1]}, "
+                                  f"errno:{-me.args[0]}, {os.strerror(-me.args[0])}")
+                        pass
+
+    def _add_snap_clone(self, track_id, snapname):
+        self.metadata_mgr.add_section("clone snaps")
+        self.metadata_mgr.update_section("clone snaps", track_id, snapname)
+        self.metadata_mgr.flush()
+
+    def _remove_snap_clone(self, track_id):
+        self.metadata_mgr.remove_option("clone snaps", track_id)
+        self.metadata_mgr.flush()
+
+    def attach_snapshot(self, snapname, tgt_subvolume):
+        if not snapname.encode('utf-8') in self.list_snapshots():
+            raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
+        try:
+            create_clone_index(self.fs, self.vol_spec)
+            with open_clone_index(self.fs, self.vol_spec) as index:
+                track_idx = index.track(tgt_subvolume.base_path)
+                self._add_snap_clone(track_idx, snapname)
+        except (IndexException, MetadataMgrException) as e:
+            log.warning("error creating clone index: {0}".format(e))
+            raise VolumeException(-errno.EINVAL, "error cloning subvolume")
+
+    def detach_snapshot(self, snapname, track_id):
+        try:
+            with open_clone_index(self.fs, self.vol_spec) as index:
+                index.untrack(track_id)
+                self._remove_snap_clone(track_id)
+        except (IndexException, MetadataMgrException) as e:
+            log.warning("error delining snapshot from clone: {0}".format(e))
+            raise VolumeException(-errno.EINVAL, "error delinking snapshot from clone")
diff --git a/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v2.py b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v2.py
new file mode 100644
index 000000000..03085d049
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/versions/subvolume_v2.py
@@ -0,0 +1,394 @@
+import os
+import stat
+import uuid
+import errno
+import logging
+
+import cephfs
+
+from .metadata_manager import MetadataManager
+from .subvolume_attrs import SubvolumeTypes, SubvolumeStates, SubvolumeFeatures
+from .op_sm import SubvolumeOpSm
+from .subvolume_v1 import SubvolumeV1
+from ..template import SubvolumeTemplate
+from ...exception import OpSmException, VolumeException, MetadataMgrException
+from ...fs_util import listdir, create_base_dir
+from ..template import SubvolumeOpType
+
+log = logging.getLogger(__name__)
+
+class SubvolumeV2(SubvolumeV1):
+    """
+    Version 2 subvolumes creates a subvolume with path as follows,
+        volumes/<group-name>/<subvolume-name>/<uuid>/
+
+    The distinguishing feature of V2 subvolume as compared to V1 subvolumes is its ability to retain snapshots
+    of a subvolume on removal. This is done by creating snapshots under the <subvolume-name> directory,
+    rather than under the <uuid> directory, as is the case of V1 subvolumes.
+
+    - The directory under which user data resides is <uuid>
+    - Snapshots of the subvolume are taken within the <subvolume-name> directory
+    - A meta file is maintained under the <subvolume-name> directory as a metadata store, storing information similar
+    to V1 subvolumes
+    - On a request to remove subvolume but retain its snapshots, only the <uuid> directory is moved to trash, retaining
+    the rest of the subvolume and its meta file.
+        - The <uuid> directory, when present, is the current incarnation of the subvolume, which may have snapshots of
+        older incarnations of the same subvolume.
+    - V1 subvolumes that currently do not have any snapshots are upgraded to V2 subvolumes automatically, to support the
+    snapshot retention feature
+    """
+    VERSION = 2
+
+    @staticmethod
+    def version():
+        return SubvolumeV2.VERSION
+
+    @property
+    def features(self):
+        return [SubvolumeFeatures.FEATURE_SNAPSHOT_CLONE.value,
+                SubvolumeFeatures.FEATURE_SNAPSHOT_AUTOPROTECT.value,
+                SubvolumeFeatures.FEATURE_SNAPSHOT_RETENTION.value]
+
+    @property
+    def retained(self):
+        try:
+            self.metadata_mgr.refresh()
+            if self.state == SubvolumeStates.STATE_RETAINED:
+                return True
+            return False
+        except MetadataMgrException as me:
+            if me.errno != -errno.ENOENT:
+                raise VolumeException(me.errno, "internal error while processing subvolume '{0}'".format(self.subvolname))
+        return False
+
+    @property
+    def purgeable(self):
+        if not self.retained or self.list_snapshots() or self.has_pending_purges:
+            return False
+        return True
+
+    @property
+    def has_pending_purges(self):
+        try:
+            return not listdir(self.fs, self.trash_dir) == []
+        except VolumeException as ve:
+            if ve.errno == -errno.ENOENT:
+                return False
+            raise
+
+    @property
+    def trash_dir(self):
+        return os.path.join(self.base_path, b".trash")
+
+    def create_trashcan(self):
+        """per subvolume trash directory"""
+        try:
+            self.fs.stat(self.trash_dir)
+        except cephfs.Error as e:
+            if e.args[0] == errno.ENOENT:
+                try:
+                    self.fs.mkdir(self.trash_dir, 0o700)
+                except cephfs.Error as ce:
+                    raise VolumeException(-ce.args[0], ce.args[1])
+            else:
+                raise VolumeException(-e.args[0], e.args[1])
+
+    def mark_subvolume(self):
+        # set subvolume attr, on subvolume root, marking it as a CephFS subvolume
+        # subvolume root is where snapshots would be taken, and hence is the base_path for v2 subvolumes
+        try:
+            # MDS treats this as a noop for already marked subvolume
+            self.fs.setxattr(self.base_path, 'ceph.dir.subvolume', b'1', 0)
+        except cephfs.InvalidValue as e:
+            raise VolumeException(-errno.EINVAL, "invalid value specified for ceph.dir.subvolume")
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    @staticmethod
+    def is_valid_uuid(uuid_str):
+        try:
+            uuid.UUID(uuid_str)
+            return True
+        except ValueError:
+            return False
+
+    def snapshot_base_path(self):
+        return os.path.join(self.base_path, self.vol_spec.snapshot_dir_prefix.encode('utf-8'))
+
+    def snapshot_data_path(self, snapname):
+        snap_base_path = self.snapshot_path(snapname)
+        uuid_str = None
+        try:
+            with self.fs.opendir(snap_base_path) as dir_handle:
+                d = self.fs.readdir(dir_handle)
+                while d:
+                    if d.d_name not in (b".", b".."):
+                        d_full_path = os.path.join(snap_base_path, d.d_name)
+                        stx = self.fs.statx(d_full_path, cephfs.CEPH_STATX_MODE, cephfs.AT_SYMLINK_NOFOLLOW)
+                        if stat.S_ISDIR(stx.get('mode')):
+                            if self.is_valid_uuid(d.d_name.decode('utf-8')):
+                                uuid_str = d.d_name
+                    d = self.fs.readdir(dir_handle)
+        except cephfs.Error as e:
+            if e.errno == errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
+            raise VolumeException(-e.args[0], e.args[1])
+
+        if not uuid_str:
+            raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
+
+        return os.path.join(snap_base_path, uuid_str)
+
+    def _remove_on_failure(self, subvol_path, retained):
+        if retained:
+            log.info("cleaning up subvolume incarnation with path: {0}".format(subvol_path))
+            try:
+                self.fs.rmdir(subvol_path)
+            except cephfs.Error as e:
+                raise VolumeException(-e.args[0], e.args[1])
+        else:
+            log.info("cleaning up subvolume with path: {0}".format(self.subvolname))
+            self.remove(internal_cleanup=True)
+
+    def _set_incarnation_metadata(self, subvolume_type, qpath, initial_state):
+        self.metadata_mgr.update_global_section(MetadataManager.GLOBAL_META_KEY_TYPE, subvolume_type.value)
+        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):
+        subvolume_type = SubvolumeTypes.TYPE_NORMAL
+        try:
+            initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
+        except OpSmException as oe:
+            raise VolumeException(-errno.EINVAL, "subvolume creation failed: internal error")
+
+        retained = self.retained
+        if retained and self.has_pending_purges:
+            raise VolumeException(-errno.EAGAIN, "asynchronous purge of subvolume in progress")
+        subvol_path = os.path.join(self.base_path, str(uuid.uuid4()).encode('utf-8'))
+        try:
+            # create group directory with default mode(0o755) if it doesn't exist.
+            create_base_dir(self.fs, self.group.path, self.vol_spec.DEFAULT_MODE)
+            self.fs.mkdirs(subvol_path, mode)
+            self.mark_subvolume()
+            attrs = {
+                'uid': uid,
+                'gid': gid,
+                'data_pool': pool,
+                'pool_namespace': self.namespace if isolate_nspace else None,
+                'quota': size
+            }
+            self.set_attrs(subvol_path, attrs)
+
+            # persist subvolume metadata
+            qpath = subvol_path.decode('utf-8')
+            if retained:
+                self._set_incarnation_metadata(subvolume_type, qpath, initial_state)
+                self.metadata_mgr.flush()
+            else:
+                self.init_config(SubvolumeV2.VERSION, subvolume_type, qpath, initial_state)
+
+            # Create the subvolume metadata file which manages auth-ids if it doesn't exist
+            self.auth_mdata_mgr.create_subvolume_metadata_file(self.group.groupname, self.subvolname)
+        except (VolumeException, MetadataMgrException, cephfs.Error) as e:
+            try:
+                self._remove_on_failure(subvol_path, retained)
+            except VolumeException as ve:
+                log.info("failed to cleanup subvolume '{0}' ({1})".format(self.subvolname, ve))
+
+            if isinstance(e, MetadataMgrException):
+                log.error("metadata manager exception: {0}".format(e))
+                e = VolumeException(-errno.EINVAL, f"exception in subvolume metadata: {os.strerror(-e.args[0])}")
+            elif isinstance(e, cephfs.Error):
+                e = VolumeException(-e.args[0], e.args[1])
+            raise e
+
+    def create_clone(self, pool, source_volname, source_subvolume, snapname):
+        subvolume_type = SubvolumeTypes.TYPE_CLONE
+        try:
+            initial_state = SubvolumeOpSm.get_init_state(subvolume_type)
+        except OpSmException as oe:
+            raise VolumeException(-errno.EINVAL, "clone failed: internal error")
+
+        retained = self.retained
+        if retained and self.has_pending_purges:
+            raise VolumeException(-errno.EAGAIN, "asynchronous purge of subvolume in progress")
+        subvol_path = os.path.join(self.base_path, str(uuid.uuid4()).encode('utf-8'))
+        try:
+            # source snapshot attrs are used to create clone subvolume
+            # attributes of subvolume's content though, are synced during the cloning process.
+            attrs = source_subvolume.get_attrs(source_subvolume.snapshot_data_path(snapname))
+
+            # The source of the clone may have exceeded its quota limit as
+            # CephFS quotas are imprecise. Cloning such a source may fail if
+            # the quota on the destination is set before starting the clone
+            # copy. So always set the quota on destination after cloning is
+            # successful.
+            attrs["quota"] = None
+
+            # override snapshot pool setting, if one is provided for the clone
+            if pool is not None:
+                attrs["data_pool"] = pool
+                attrs["pool_namespace"] = None
+
+            # create directory and set attributes
+            self.fs.mkdirs(subvol_path, attrs.get("mode"))
+            self.mark_subvolume()
+            self.set_attrs(subvol_path, attrs)
+
+            # persist subvolume metadata and clone source
+            qpath = subvol_path.decode('utf-8')
+            if retained:
+                self._set_incarnation_metadata(subvolume_type, qpath, initial_state)
+            else:
+                self.metadata_mgr.init(SubvolumeV2.VERSION, subvolume_type.value, qpath, initial_state.value)
+            self.add_clone_source(source_volname, source_subvolume, snapname)
+            self.metadata_mgr.flush()
+        except (VolumeException, MetadataMgrException, cephfs.Error) as e:
+            try:
+                self._remove_on_failure(subvol_path, retained)
+            except VolumeException as ve:
+                log.info("failed to cleanup subvolume '{0}' ({1})".format(self.subvolname, ve))
+
+            if isinstance(e, MetadataMgrException):
+                log.error("metadata manager exception: {0}".format(e))
+                e = VolumeException(-errno.EINVAL, f"exception in subvolume metadata: {os.strerror(-e.args[0])}")
+            elif isinstance(e, cephfs.Error):
+                e = VolumeException(-e.args[0], e.args[1])
+            raise e
+
+    def allowed_ops_by_type(self, vol_type):
+        if vol_type == SubvolumeTypes.TYPE_CLONE:
+            return {op_type for op_type in SubvolumeOpType}
+
+        if vol_type == SubvolumeTypes.TYPE_NORMAL:
+            return {op_type for op_type in SubvolumeOpType} - {SubvolumeOpType.CLONE_STATUS,
+                                                               SubvolumeOpType.CLONE_CANCEL,
+                                                               SubvolumeOpType.CLONE_INTERNAL}
+
+        return {}
+
+    def allowed_ops_by_state(self, vol_state):
+        if vol_state == SubvolumeStates.STATE_COMPLETE:
+            return {op_type for op_type in SubvolumeOpType}
+
+        if vol_state == SubvolumeStates.STATE_RETAINED:
+            return {
+                SubvolumeOpType.REMOVE,
+                SubvolumeOpType.REMOVE_FORCE,
+                SubvolumeOpType.LIST,
+                SubvolumeOpType.INFO,
+                SubvolumeOpType.SNAP_REMOVE,
+                SubvolumeOpType.SNAP_LIST,
+                SubvolumeOpType.SNAP_INFO,
+                SubvolumeOpType.SNAP_PROTECT,
+                SubvolumeOpType.SNAP_UNPROTECT,
+                SubvolumeOpType.CLONE_SOURCE
+            }
+
+        return {SubvolumeOpType.REMOVE_FORCE,
+                SubvolumeOpType.CLONE_CREATE,
+                SubvolumeOpType.CLONE_STATUS,
+                SubvolumeOpType.CLONE_CANCEL,
+                SubvolumeOpType.CLONE_INTERNAL,
+                SubvolumeOpType.CLONE_SOURCE}
+
+    def open(self, op_type):
+        if not isinstance(op_type, SubvolumeOpType):
+            raise VolumeException(-errno.ENOTSUP, "operation {0} not supported on subvolume '{1}'".format(
+                                  op_type.value, self.subvolname))
+        try:
+            self.metadata_mgr.refresh()
+            # unconditionally mark as subvolume, to handle pre-existing subvolumes without the mark
+            self.mark_subvolume()
+
+            etype = self.subvol_type
+            if op_type not in self.allowed_ops_by_type(etype):
+                raise VolumeException(-errno.ENOTSUP, "operation '{0}' is not allowed on subvolume '{1}' of type {2}".format(
+                                      op_type.value, self.subvolname, etype.value))
+
+            estate = self.state
+            if op_type not in self.allowed_ops_by_state(estate) and estate == SubvolumeStates.STATE_RETAINED:
+                raise VolumeException(-errno.ENOENT, "subvolume '{0}' is removed and has only snapshots retained".format(
+                                      self.subvolname))
+
+            if op_type not in self.allowed_ops_by_state(estate) and estate != SubvolumeStates.STATE_RETAINED:
+                raise VolumeException(-errno.EAGAIN, "subvolume '{0}' is not ready for operation {1}".format(
+                                      self.subvolname, op_type.value))
+
+            if estate != SubvolumeStates.STATE_RETAINED:
+                subvol_path = self.path
+                log.debug("refreshed metadata, checking subvolume path '{0}'".format(subvol_path))
+                st = self.fs.stat(subvol_path)
+
+                self.uid = int(st.st_uid)
+                self.gid = int(st.st_gid)
+                self.mode = int(st.st_mode & ~stat.S_IFMT(st.st_mode))
+        except MetadataMgrException as me:
+            if me.errno == -errno.ENOENT:
+                raise VolumeException(-errno.ENOENT, "subvolume '{0}' does not exist".format(self.subvolname))
+            raise VolumeException(me.args[0], me.args[1])
+        except cephfs.ObjectNotFound:
+            log.debug("missing subvolume path '{0}' for subvolume '{1}'".format(subvol_path, self.subvolname))
+            raise VolumeException(-errno.ENOENT, "mount path missing for subvolume '{0}'".format(self.subvolname))
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    def trash_incarnation_dir(self):
+        """rename subvolume (uuid component) to trash"""
+        self.create_trashcan()
+        try:
+            bname = os.path.basename(self.path)
+            tpath = os.path.join(self.trash_dir, bname)
+            log.debug("trash: {0} -> {1}".format(self.path, tpath))
+            self.fs.rename(self.path, tpath)
+            self._link_dir(tpath, bname)
+        except cephfs.Error as e:
+            raise VolumeException(-e.args[0], e.args[1])
+
+    @staticmethod
+    def safe_to_remove_subvolume_clone(subvol_state):
+        # Both the STATE_FAILED and STATE_CANCELED are handled by 'handle_clone_failed' in the state
+        # machine which removes the entry from the index. Hence, it's safe to removed clone with
+        # force option for both.
+        acceptable_rm_clone_states = [SubvolumeStates.STATE_COMPLETE, SubvolumeStates.STATE_CANCELED,
+                                      SubvolumeStates.STATE_FAILED, SubvolumeStates.STATE_RETAINED]
+        if subvol_state not in acceptable_rm_clone_states:
+            return False
+        return True
+
+    def remove(self, retainsnaps=False, internal_cleanup=False):
+        if self.list_snapshots():
+            if not retainsnaps:
+                raise VolumeException(-errno.ENOTEMPTY, "subvolume '{0}' has snapshots".format(self.subvolname))
+        else:
+            if not internal_cleanup and not self.safe_to_remove_subvolume_clone(self.state):
+                raise VolumeException(-errno.EAGAIN,
+                                      "{0} clone in-progress -- please cancel the clone and retry".format(self.subvolname))
+            if not self.has_pending_purges:
+                self.trash_base_dir()
+                # Delete the volume meta file, if it's not already deleted
+                self.auth_mdata_mgr.delete_subvolume_metadata_file(self.group.groupname, self.subvolname)
+                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()
+            # Delete the volume meta file, if it's not already deleted
+            self.auth_mdata_mgr.delete_subvolume_metadata_file(self.group.groupname, self.subvolname)
+
+    def info(self):
+        if self.state != SubvolumeStates.STATE_RETAINED:
+            return super(SubvolumeV2, self).info()
+
+        return {'type': self.subvol_type.value, 'features': self.features, 'state': SubvolumeStates.STATE_RETAINED.value}
+
+    def remove_snapshot(self, snapname, force=False):
+        super(SubvolumeV2, self).remove_snapshot(snapname, force)
+        if self.purgeable:
+            self.trash_base_dir()
+            # tickle the volume purge job to purge this entry, using ESTALE
+            raise VolumeException(-errno.ESTALE, "subvolume '{0}' has been removed as the last retained snapshot is removed".format(self.subvolname))
+        # if not purgeable, subvol is not retained, or has snapshots, or already has purge jobs that will garbage collect this subvol
diff --git a/src/pybind/mgr/volumes/fs/operations/volume.py b/src/pybind/mgr/volumes/fs/operations/volume.py
new file mode 100644
index 000000000..a79aa55e1
--- /dev/null
+++ b/src/pybind/mgr/volumes/fs/operations/volume.py
@@ -0,0 +1,198 @@
+import errno
+import logging
+import os
+
+from typing import List
+
+from contextlib import contextmanager
+
+import orchestrator
+
+from .lock import GlobalLock
+from ..exception import VolumeException
+from ..fs_util import create_pool, remove_pool, create_filesystem, \
+    remove_filesystem, create_mds, volume_exists
+from .trash import Trash
+from mgr_util import open_filesystem, CephfsConnectionException
+
+log = logging.getLogger(__name__)
+
+def gen_pool_names(volname):
+    """
+    return metadata and data pool name (from a filesystem/volume name) as a tuple
+    """
+    return "cephfs.{}.meta".format(volname), "cephfs.{}.data".format(volname)
+
+def get_mds_map(mgr, volname):
+    """
+    return mdsmap for a volname
+    """
+    mds_map = None
+    fs_map = mgr.get("fs_map")
+    for f in fs_map['filesystems']:
+        if volname == f['mdsmap']['fs_name']:
+            return f['mdsmap']
+    return mds_map
+
+def get_pool_names(mgr, volname):
+    """
+    return metadata and data pools (list) names of volume as a tuple
+    """
+    fs_map = mgr.get("fs_map")
+    metadata_pool_id = None
+    data_pool_ids = [] # type: List[int]
+    for f in fs_map['filesystems']:
+        if volname == f['mdsmap']['fs_name']:
+            metadata_pool_id = f['mdsmap']['metadata_pool']
+            data_pool_ids = f['mdsmap']['data_pools']
+            break
+    if metadata_pool_id is None:
+        return None, None
+
+    osdmap = mgr.get("osd_map")
+    pools = dict([(p['pool'], p['pool_name']) for p in osdmap['pools']])
+    metadata_pool = pools[metadata_pool_id]
+    data_pools = [pools[id] for id in data_pool_ids]
+    return metadata_pool, data_pools
+
+def get_pool_ids(mgr, volname):
+    """
+    return metadata and data pools (list) id of volume as a tuple
+    """
+    fs_map = mgr.get("fs_map")
+    metadata_pool_id = None
+    data_pool_ids = [] # type: List[int]
+    for f in fs_map['filesystems']:
+        if volname == f['mdsmap']['fs_name']:
+            metadata_pool_id = f['mdsmap']['metadata_pool']
+            data_pool_ids = f['mdsmap']['data_pools']
+            break
+    if metadata_pool_id is None:
+        return None, None
+    return metadata_pool_id, data_pool_ids
+
+def create_volume(mgr, volname, placement):
+    """
+    create volume  (pool, filesystem and mds)
+    """
+    metadata_pool, data_pool = gen_pool_names(volname)
+    # create pools
+    r, outb, outs = create_pool(mgr, metadata_pool)
+    if r != 0:
+        return r, outb, outs
+    r, outb, outs = create_pool(mgr, data_pool)
+    if r != 0:
+        #cleanup
+        remove_pool(mgr, metadata_pool)
+        return r, outb, outs
+    # create filesystem
+    r, outb, outs = create_filesystem(mgr, volname, metadata_pool, data_pool)
+    if r != 0:
+        log.error("Filesystem creation error: {0} {1} {2}".format(r, outb, outs))
+        #cleanup
+        remove_pool(mgr, data_pool)
+        remove_pool(mgr, metadata_pool)
+        return r, outb, outs
+    return create_mds(mgr, volname, placement)
+
+
+def delete_volume(mgr, volname, metadata_pool, data_pools):
+    """
+    delete the given module (tear down mds, remove filesystem, remove pools)
+    """
+    # Tear down MDS daemons
+    try:
+        completion = mgr.remove_service('mds.' + volname)
+        orchestrator.raise_if_exception(completion)
+    except (ImportError, orchestrator.OrchestratorError):
+        log.warning("OrchestratorError, not tearing down MDS daemons")
+    except Exception as e:
+        # Don't let detailed orchestrator exceptions (python backtraces)
+        # bubble out to the user
+        log.exception("Failed to tear down MDS daemons")
+        return -errno.EINVAL, "", str(e)
+
+    # In case orchestrator didn't tear down MDS daemons cleanly, or
+    # there was no orchestrator, we force the daemons down.
+    if volume_exists(mgr, volname):
+        r, outb, outs = remove_filesystem(mgr, volname)
+        if r != 0:
+            return r, outb, outs
+    else:
+        err = "Filesystem not found for volume '{0}'".format(volname)
+        log.warning(err)
+        return -errno.ENOENT, "", err
+    r, outb, outs = remove_pool(mgr, metadata_pool)
+    if r != 0:
+        return r, outb, outs
+
+    for data_pool in data_pools:
+        r, outb, outs = remove_pool(mgr, data_pool)
+        if r != 0:
+            return r, outb, outs
+    result_str = "metadata pool: {0} data pool: {1} removed".format(metadata_pool, str(data_pools))
+    return r, result_str, ""
+
+
+def list_volumes(mgr):
+    """
+    list all filesystem volumes.
+
+    :param: None
+    :return: None
+    """
+    result = []
+    fs_map = mgr.get("fs_map")
+    for f in fs_map['filesystems']:
+        result.append({'name': f['mdsmap']['fs_name']})
+    return result
+
+
+def get_pending_subvol_deletions_count(path):
+    """
+    Get the number of pending subvolumes deletions.
+    """
+    trashdir = os.path.join(path, Trash.GROUP_NAME)
+    try:
+        num_pending_subvol_del = len(os.listdir(trashdir))
+    except OSError as e:
+        if e.errno == errno.ENOENT:
+            num_pending_subvol_del = 0
+
+    return {'pending_subvolume_deletions': num_pending_subvol_del}
+
+
+@contextmanager
+def open_volume(vc, volname):
+    """
+    open a volume for exclusive access. This API is to be used as a contextr
+    manager.
+
+    :param vc: volume client instance
+    :param volname: volume name
+    :return: yields a volume handle (ceph filesystem handle)
+    """
+    g_lock = GlobalLock()
+    with g_lock.lock_op():
+        try:
+            with open_filesystem(vc, volname) as fs_handle:
+                yield fs_handle
+        except CephfsConnectionException as ce:
+            raise VolumeException(ce.errno, ce.error_str)
+
+
+@contextmanager
+def open_volume_lockless(vc, volname):
+    """
+    open a volume with shared access. This API is to be used as a context
+    manager.
+
+    :param vc: volume client instance
+    :param volname: volume name
+    :return: yields a volume handle (ceph filesystem handle)
+    """
+    try:
+        with open_filesystem(vc, volname) as fs_handle:
+            yield fs_handle
+    except CephfsConnectionException as ce:
+        raise VolumeException(ce.errno, ce.error_str)