Repository: ceph/go-ceph
Pull request #372 (2020-09-24 21:01:57)

cephfs admin: subvolume clone support
Clone support for cephfs subvolumes.

It comes with a response handling refactor and the addition of Force remove methods.

As part of cloning I had to deal with numerous differences between ceph versions and these are inter-codename differences. IOW, some "nautilus" releases behave one way, some newer "nautilus" release behave another way. At the time of this writing some changes have been back-ported into nautilus but not yet into octopus.
This is all to say that various changes/refactoring/whatnot that made this a bigger PR are due to issues found testing and working with what's ended up being the most complicated part of this module.

## Checklist
- [x] Added tests for features and functional changes
- [x] Public functions and types are documented
- [x] Standard formatting is applied to Go code


phlogistonjohn: 
This nearly completes the cephfs admin package but there's still some more practical testing and additional cleanups that need to be done. But we're getting close!
phlogistonjohn: Updated.

Patch:
diff --git a/cephfs/admin/clone.go b/cephfs/admin/clone.go
new file mode 100644
index 0000000..9f6c4f2
--- /dev/null
+++ b/cephfs/admin/clone.go
@@ -0,0 +1,132 @@
+// +build !luminous,!mimic
+
+package admin
+
+import (
+	"strings"
+)
+
+const notProtectedSuffix = "is not protected"
+
+// NotProtectedError error values will be returned by CloneSubVolumeSnapshot in
+// the case that the source snapshot needs to be protected but is not.  The
+// requirement for a snapshot to be protected prior to cloning varies by Ceph
+// version.
+type NotProtectedError struct {
+	response
+}
+
+// CloneOptions are used to specify optional values to be used when creating a
+// new subvolume clone.
+type CloneOptions struct {
+	TargetGroup string
+	PoolLayout  string
+}
+
+// CloneSubVolumeSnapshot Clones the specified snapshot from the subvolume.
+//
+// Similar To:
+//  ceph fs subvolume snapshot clone <volume> <subvolume> <snapshot> <name>
+func (fsa *FSAdmin) CloneSubVolumeSnapshot(volume, group, subvolume, snapshot, name string, o *CloneOptions) error {
+	m := map[string]string{
+		"prefix":          "fs subvolume snapshot clone",
+		"vol_name":        volume,
+		"sub_name":        subvolume,
+		"snap_name":       snapshot,
+		"target_sub_name": name,
+		"format":          "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	if o != nil && o.TargetGroup != NoGroup {
+		m["target_group_name"] = group
+	}
+	if o != nil && o.PoolLayout != "" {
+		m["pool_layout"] = o.PoolLayout
+	}
+	return checkCloneResponse(fsa.marshalMgrCommand(m))
+}
+
+func checkCloneResponse(res response) error {
+	if strings.HasSuffix(res.status, notProtectedSuffix) {
+		return NotProtectedError{response: res}
+	}
+	return res.noData().End()
+}
+
+// CloneState is used to define constant values used to determine the state of
+// a clone.
+type CloneState string
+
+const (
+	// ClonePending is the state of a pending clone.
+	ClonePending = CloneState("pending")
+	// CloneInProgress is the state of a clone in progress.
+	CloneInProgress = CloneState("in-progress")
+	// CloneComplete is the state of a complete clone.
+	CloneComplete = CloneState("complete")
+	// CloneFailed is the state of a failed clone.
+	CloneFailed = CloneState("failed")
+)
+
+// CloneSource contains values indicating the source of a clone.
+type CloneSource struct {
+	Volume    string `json:"volume"`
+	Group     string `json:"group"`
+	SubVolume string `json:"subvolume"`
+	Snapshot  string `json:"snapshot"`
+}
+
+// CloneStatus reports on the status of a subvolume clone.
+type CloneStatus struct {
+	State  CloneState  `json:"state"`
+	Source CloneSource `json:"source"`
+}
+
+type cloneStatusWrapper struct {
+	Status CloneStatus `json:"status"`
+}
+
+func parseCloneStatus(res response) (*CloneStatus, error) {
+	var status cloneStatusWrapper
+	if err := res.noStatus().unmarshal(&status).End(); err != nil {
+		return nil, err
+	}
+	return &status.Status, nil
+}
+
+// CloneStatus returns data reporting the status of a subvolume clone.
+//
+// Similar To:
+//  ceph fs clone status <volume> --group_name=<group> <clone>
+func (fsa *FSAdmin) CloneStatus(volume, group, clone string) (*CloneStatus, error) {
+	m := map[string]string{
+		"prefix":     "fs clone status",
+		"vol_name":   volume,
+		"clone_name": clone,
+		"format":     "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return parseCloneStatus(fsa.marshalMgrCommand(m))
+}
+
+// CancelClone stops the background processes that populate a clone.
+// CancelClone does not delete the clone.
+//
+// Similar To:
+//  ceph fs clone cancel <volume> --group_name=<group> <clone>
+func (fsa *FSAdmin) CancelClone(volume, group, clone string) error {
+	m := map[string]string{
+		"prefix":     "fs clone cancel",
+		"vol_name":   volume,
+		"clone_name": clone,
+		"format":     "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return fsa.marshalMgrCommand(m).noData().End()
+}
diff --git a/cephfs/admin/fsadmin.go b/cephfs/admin/fsadmin.go
index 1312786..2b49980 100644
--- a/cephfs/admin/fsadmin.go
+++ b/cephfs/admin/fsadmin.go
@@ -4,7 +4,6 @@ package admin
 
 import (
 	"encoding/json"
-	"fmt"
 	"strconv"
 
 	"github.com/ceph/go-ceph/rados"
@@ -60,38 +59,38 @@ func (fsa *FSAdmin) validate() error {
 
 // rawMgrCommand takes a byte buffer and sends it to the MGR as a command.
 // The buffer is expected to contain preformatted JSON.
-func (fsa *FSAdmin) rawMgrCommand(buf []byte) ([]byte, string, error) {
+func (fsa *FSAdmin) rawMgrCommand(buf []byte) response {
 	if err := fsa.validate(); err != nil {
-		return nil, "", err
+		return response{err: err}
 	}
-	return fsa.conn.MgrCommand([][]byte{buf})
+	return newResponse(fsa.conn.MgrCommand([][]byte{buf}))
 }
 
 // marshalMgrCommand takes an generic interface{} value, converts it to JSON and
 // sends the json to the MGR as a command.
-func (fsa *FSAdmin) marshalMgrCommand(v interface{}) ([]byte, string, error) {
+func (fsa *FSAdmin) marshalMgrCommand(v interface{}) response {
 	b, err := json.Marshal(v)
 	if err != nil {
-		return nil, "", err
+		return response{err: err}
 	}
 	return fsa.rawMgrCommand(b)
 }
 
 // rawMonCommand takes a byte buffer and sends it to the MON as a command.
 // The buffer is expected to contain preformatted JSON.
-func (fsa *FSAdmin) rawMonCommand(buf []byte) ([]byte, string, error) {
+func (fsa *FSAdmin) rawMonCommand(buf []byte) response {
 	if err := fsa.validate(); err != nil {
-		return nil, "", err
+		return response{err: err}
 	}
-	return fsa.conn.MonCommand(buf)
+	return newResponse(fsa.conn.MonCommand(buf))
 }
 
 // marshalMonCommand takes an generic interface{} value, converts it to JSON and
 // sends the json to the MGR as a command.
-func (fsa *FSAdmin) marshalMonCommand(v interface{}) ([]byte, string, error) {
+func (fsa *FSAdmin) marshalMonCommand(v interface{}) response {
 	b, err := json.Marshal(v)
 	if err != nil {
-		return nil, "", err
+		return response{err: err}
 	}
 	return fsa.rawMonCommand(b)
 }
@@ -100,9 +99,9 @@ type listNamedResult struct {
 	Name string `json:"name"`
 }
 
-func parseListNames(res []byte, status string, err error) ([]string, error) {
+func parseListNames(res response) ([]string, error) {
 	var r []listNamedResult
-	if err := unmarshalResponseJSON(res, status, err, &r); err != nil {
+	if err := res.noStatus().unmarshal(&r).End(); err != nil {
 		return nil, err
 	}
 	vl := make([]string, len(r))
@@ -112,47 +111,20 @@ func parseListNames(res []byte, status string, err error) ([]string, error) {
 	return vl, nil
 }
 
-// checkEmptyResponseExpected returns an error if the result or status
-// are non-empty.
-func checkEmptyResponseExpected(res []byte, status string, err error) error {
-	if err != nil {
-		return err
-	}
-	if len(res) != 0 {
-		return fmt.Errorf("unexpected response: %s", string(res))
-	}
-	if status != "" {
-		return fmt.Errorf("error status: %s", status)
-	}
-	return nil
-}
-
-func unmarshalResponseJSON(res []byte, status string, err error, v interface{}) error {
-	if err != nil {
-		return err
-	}
-	if status != "" {
-		return fmt.Errorf("error status: %s", status)
-	}
-	return json.Unmarshal(res, v)
-}
-
-// extractPathResponse returns a cleaned up path from requests that get a path
+// parsePathResponse returns a cleaned up path from requests that get a path
 // unless an error is encountered, then an error is returned.
-func extractPathResponse(res []byte, status string, err error) (string, error) {
-	if err != nil {
-		return "", err
-	}
-	if status != "" {
-		return "", fmt.Errorf("error status: %s", status)
+func parsePathResponse(res response) (string, error) {
+	if res2 := res.noStatus(); !res2.Ok() {
+		return "", res.End()
 	}
+	b := res.body
 	// if there's a trailing newline in the buffer strip it.
 	// ceph assumes a CLI wants the output of the buffer and there's
 	// no format=json mode available currently.
-	for len(res) >= 1 && res[len(res)-1] == '\n' {
-		res = res[:len(res)-1]
+	for len(b) >= 1 && b[len(b)-1] == '\n' {
+		b = b[:len(b)-1]
 	}
-	return string(res), nil
+	return string(b), nil
 }
 
 // modeString converts a unix-style mode value to a string-ified version in an
@@ -171,3 +143,18 @@ func modeString(m int, force bool) string {
 func uint64String(v uint64) string {
 	return strconv.FormatUint(uint64(v), 10)
 }
+
+type rmFlags struct {
+	force bool
+}
+
+func (f rmFlags) Update(m map[string]string) map[string]interface{} {
+	o := make(map[string]interface{})
+	for k, v := range m {
+		o[k] = v
+	}
+	if f.force {
+		o["force"] = true
+	}
+	return o
+}
diff --git a/cephfs/admin/response.go b/cephfs/admin/response.go
new file mode 100644
index 0000000..fb0e407
--- /dev/null
+++ b/cephfs/admin/response.go
@@ -0,0 +1,119 @@
+// +build !luminous,!mimic
+
+package admin
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+var (
+	// ErrStatusNotEmpty may be returned if a call should not have a status
+	// string set but one is.
+	ErrStatusNotEmpty = errors.New("response status not empty")
+	// ErrBodyNotEmpty may be returned if a call should have an empty body but
+	// a body value is present.
+	ErrBodyNotEmpty = errors.New("response body not empty")
+)
+
+const (
+	deprecatedSuffix = "call is deprecated and will be removed in a future release"
+)
+
+// response encapsulates the data returned by ceph and supports easy processing
+// pipelines.
+type response struct {
+	body   []byte
+	status string
+	err    error
+}
+
+// Ok returns true if the response contains no error.
+func (r response) Ok() bool {
+	return r.err == nil
+}
+
+// Error implements the error interface.
+func (r response) Error() string {
+	if r.status == "" {
+		return r.err.Error()
+	}
+	return fmt.Sprintf("%s: %q", r.err, r.status)
+}
+
+// Unwrap returns the error this response contains.
+func (r response) Unwrap() error {
+	return r.err
+}
+
+// Status returns the status string value.
+func (r response) Status() string {
+	return r.status
+}
+
+// End returns an error if the response contains an error or nil, indicating
+// that response is no longer needed for processing.
+func (r response) End() error {
+	if !r.Ok() {
+		return r
+	}
+	return nil
+}
+
+// noStatus asserts that the input response has no status value.
+func (r response) noStatus() response {
+	if !r.Ok() {
+		return r
+	}
+	if r.status != "" {
+		return response{r.body, r.status, ErrStatusNotEmpty}
+	}
+	return r
+}
+
+// noBody asserts that the input response has no body value.
+func (r response) noBody() response {
+	if !r.Ok() {
+		return r
+	}
+	if len(r.body) != 0 {
+		return response{r.body, r.status, ErrBodyNotEmpty}
+	}
+	return r
+}
+
+// noData asserts that the input response has no status or body values.
+func (r response) noData() response {
+	return r.noStatus().noBody()
+}
+
+// filterDeprecated removes deprecation warnings from the response status.
+// Use it when checking the response from calls that may be deprecated in ceph
+// if you want those calls to continue working if the warning is present.
+func (r response) filterDeprecated() response {
+	if !r.Ok() {
+		return r
+	}
+	if strings.HasSuffix(r.status, deprecatedSuffix) {
+		return response{r.body, "", r.err}
+	}
+	return r
+}
+
+// unmarshal data from the response body into v.
+func (r response) unmarshal(v interface{}) response {
+	if !r.Ok() {
+		return r
+	}
+	if err := json.Unmarshal(r.body, v); err != nil {
+		return response{body: r.body, err: err}
+	}
+	return r
+}
+
+// newResponse returns a response.
+func newResponse(b []byte, s string, e error) response {
+	return response{b, s, e}
+}
diff --git a/cephfs/admin/subvolume.go b/cephfs/admin/subvolume.go
index ac3cb90..6a0c2c2 100644
--- a/cephfs/admin/subvolume.go
+++ b/cephfs/admin/subvolume.go
@@ -61,7 +61,7 @@ func (fsa *FSAdmin) CreateSubVolume(volume, group, name string, o *SubVolumeOpti
 		o = &SubVolumeOptions{}
 	}
 	f := o.toFields(volume, group, name)
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(f))
+	return fsa.marshalMgrCommand(f).noData().End()
 }
 
 // ListSubVolumes returns a list of subvolumes belonging to the volume and
@@ -85,8 +85,21 @@ func (fsa *FSAdmin) ListSubVolumes(volume, group string) ([]string, error) {
 // subvolume group.
 //
 // Similar To:
-//  ceph fs subvolume rm <volume> --group-name=<group> <name> ...
+//  ceph fs subvolume rm <volume> --group-name=<group> <name>
 func (fsa *FSAdmin) RemoveSubVolume(volume, group, name string) error {
+	return fsa.rmSubVolume(volume, group, name, rmFlags{})
+}
+
+// ForceRemoveSubVolume will delete a CephFS subvolume in a volume and optional
+// subvolume group.
+//
+// Similar To:
+//  ceph fs subvolume rm <volume> --group-name=<group> <name> --force
+func (fsa *FSAdmin) ForceRemoveSubVolume(volume, group, name string) error {
+	return fsa.rmSubVolume(volume, group, name, rmFlags{force: true})
+}
+
+func (fsa *FSAdmin) rmSubVolume(volume, group, name string, o rmFlags) error {
 	m := map[string]string{
 		"prefix":   "fs subvolume rm",
 		"vol_name": volume,
@@ -96,7 +109,7 @@ func (fsa *FSAdmin) RemoveSubVolume(volume, group, name string) error {
 	if group != NoGroup {
 		m["group_name"] = group
 	}
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+	return fsa.marshalMgrCommand(o.Update(m)).noData().End()
 }
 
 type subVolumeResizeFields struct {
@@ -137,8 +150,8 @@ func (fsa *FSAdmin) ResizeSubVolume(
 		NoShrink:  noShrink,
 	}
 	var result []*SubVolumeResizeResult
-	r, s, err := fsa.marshalMgrCommand(f)
-	if err := unmarshalResponseJSON(r, s, err, &result); err != nil {
+	res := fsa.marshalMgrCommand(f)
+	if err := res.noStatus().unmarshal(&result).End(); err != nil {
 		return nil, err
 	}
 	return result[0], nil
@@ -158,7 +171,7 @@ func (fsa *FSAdmin) SubVolumePath(volume, group, name string) (string, error) {
 	if group != NoGroup {
 		m["group_name"] = group
 	}
-	return extractPathResponse(fsa.marshalMgrCommand(m))
+	return parsePathResponse(fsa.marshalMgrCommand(m))
 }
 
 // SubVolumeInfo reports various informational values about a subvolume.
@@ -184,9 +197,9 @@ type subVolumeInfoWrapper struct {
 	VBytesQuota *quotaSizePlaceholder `json:"bytes_quota"`
 }
 
-func parseSubVolumeInfo(r []byte, s string, err error) (*SubVolumeInfo, error) {
+func parseSubVolumeInfo(res response) (*SubVolumeInfo, error) {
 	var info subVolumeInfoWrapper
-	if err := unmarshalResponseJSON(r, s, err, &info); err != nil {
+	if err := res.noStatus().unmarshal(&info).End(); err != nil {
 		return nil, err
 	}
 	if info.VBytesQuota != nil {
@@ -227,7 +240,7 @@ func (fsa *FSAdmin) CreateSubVolumeSnapshot(volume, group, source, name string)
 	if group != NoGroup {
 		m["group_name"] = group
 	}
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+	return fsa.marshalMgrCommand(m).noData().End()
 }
 
 // RemoveSubVolumeSnapshot removes the specified snapshot from the subvolume.
@@ -235,6 +248,19 @@ func (fsa *FSAdmin) CreateSubVolumeSnapshot(volume, group, source, name string)
 // Similar To:
 //  ceph fs subvolume snapshot rm <volume> --group-name=<group> <subvolume> <name>
 func (fsa *FSAdmin) RemoveSubVolumeSnapshot(volume, group, subvolume, name string) error {
+	return fsa.rmSubVolumeSnapshot(volume, group, subvolume, name, rmFlags{})
+}
+
+// ForceRemoveSubVolumeSnapshot removes the specified snapshot from the subvolume.
+//
+// Similar To:
+//  ceph fs subvolume snapshot rm <volume> --group-name=<group> <subvolume> <name> --force
+func (fsa *FSAdmin) ForceRemoveSubVolumeSnapshot(volume, group, subvolume, name string) error {
+	return fsa.rmSubVolumeSnapshot(volume, group, subvolume, name, rmFlags{force: true})
+}
+
+func (fsa *FSAdmin) rmSubVolumeSnapshot(volume, group, subvolume, name string, o rmFlags) error {
+
 	m := map[string]string{
 		"prefix":    "fs subvolume snapshot rm",
 		"vol_name":  volume,
@@ -245,7 +271,7 @@ func (fsa *FSAdmin) RemoveSubVolumeSnapshot(volume, group, subvolume, name strin
 	if group != NoGroup {
 		m["group_name"] = group
 	}
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+	return fsa.marshalMgrCommand(o.Update(m)).noData().End()
 }
 
 // ListSubVolumeSnapshots returns a listing of snapshots for a given subvolume.
@@ -264,3 +290,39 @@ func (fsa *FSAdmin) ListSubVolumeSnapshots(volume, group, name string) ([]string
 	}
 	return parseListNames(fsa.marshalMgrCommand(m))
 }
+
+// ProtectSubVolumeSnapshot protects the specified snapshot.
+//
+// Similar To:
+//  ceph fs subvolume snapshot protect <volume> --group-name=<group> <subvolume> <name>
+func (fsa *FSAdmin) ProtectSubVolumeSnapshot(volume, group, subvolume, name string) error {
+	m := map[string]string{
+		"prefix":    "fs subvolume snapshot protect",
+		"vol_name":  volume,
+		"sub_name":  subvolume,
+		"snap_name": name,
+		"format":    "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return fsa.marshalMgrCommand(m).filterDeprecated().noData().End()
+}
+
+// UnprotectSubVolumeSnapshot removes protection from the specified snapshot.
+//
+// Similar To:
+//  ceph fs subvolume snapshot unprotect <volume> --group-name=<group> <subvolume> <name>
+func (fsa *FSAdmin) UnprotectSubVolumeSnapshot(volume, group, subvolume, name string) error {
+	m := map[string]string{
+		"prefix":    "fs subvolume snapshot unprotect",
+		"vol_name":  volume,
+		"sub_name":  subvolume,
+		"snap_name": name,
+		"format":    "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return fsa.marshalMgrCommand(m).filterDeprecated().noData().End()
+}
diff --git a/cephfs/admin/subvolume_octopus.go b/cephfs/admin/subvolume_octopus.go
index 615712b..afb6ca7 100644
--- a/cephfs/admin/subvolume_octopus.go
+++ b/cephfs/admin/subvolume_octopus.go
@@ -11,9 +11,9 @@ type SubVolumeSnapshotInfo struct {
 	Size             ByteCount `json:"size"`
 }
 
-func parseSubVolumeSnapshotInfo(r []byte, s string, err error) (*SubVolumeSnapshotInfo, error) {
+func parseSubVolumeSnapshotInfo(res response) (*SubVolumeSnapshotInfo, error) {
 	var info SubVolumeSnapshotInfo
-	if err := unmarshalResponseJSON(r, s, err, &info); err != nil {
+	if err := res.noStatus().unmarshal(&info).End(); err != nil {
 		return nil, err
 	}
 	return &info, nil
diff --git a/cephfs/admin/subvolumegroup.go b/cephfs/admin/subvolumegroup.go
index 03b77b2..f5426ba 100644
--- a/cephfs/admin/subvolumegroup.go
+++ b/cephfs/admin/subvolumegroup.go
@@ -47,8 +47,8 @@ func (fsa *FSAdmin) CreateSubVolumeGroup(volume, name string, o *SubVolumeGroupO
 	if o == nil {
 		o = &SubVolumeGroupOptions{}
 	}
-	r, s, err := fsa.marshalMgrCommand(o.toFields(volume, name))
-	return checkEmptyResponseExpected(r, s, err)
+	res := fsa.marshalMgrCommand(o.toFields(volume, name))
+	return res.noData().End()
 }
 
 // ListSubVolumeGroups returns a list of subvolume groups belonging to the
@@ -57,25 +57,36 @@ func (fsa *FSAdmin) CreateSubVolumeGroup(volume, name string, o *SubVolumeGroupO
 // Similar To:
 //  ceph fs subvolumegroup ls cephfs <volume>
 func (fsa *FSAdmin) ListSubVolumeGroups(volume string) ([]string, error) {
-	r, s, err := fsa.marshalMgrCommand(map[string]string{
+	res := fsa.marshalMgrCommand(map[string]string{
 		"prefix":   "fs subvolumegroup ls",
 		"vol_name": volume,
 		"format":   "json",
 	})
-	return parseListNames(r, s, err)
+	return parseListNames(res)
 }
 
 // RemoveSubVolumeGroup will delete a subvolume group in a volume.
 // Similar To:
 //  ceph fs subvolumegroup rm <volume> <group_name>
 func (fsa *FSAdmin) RemoveSubVolumeGroup(volume, name string) error {
-	r, s, err := fsa.marshalMgrCommand(map[string]string{
+	return fsa.rmSubVolumeGroup(volume, name, rmFlags{})
+}
+
+// ForceRemoveSubVolumeGroup will delete a subvolume group in a volume.
+// Similar To:
+//  ceph fs subvolumegroup rm <volume> <group_name> --force
+func (fsa *FSAdmin) ForceRemoveSubVolumeGroup(volume, name string) error {
+	return fsa.rmSubVolumeGroup(volume, name, rmFlags{force: true})
+}
+
+func (fsa *FSAdmin) rmSubVolumeGroup(volume, name string, o rmFlags) error {
+	res := fsa.marshalMgrCommand(o.Update(map[string]string{
 		"prefix":     "fs subvolumegroup rm",
 		"vol_name":   volume,
 		"group_name": name,
 		"format":     "json",
-	})
-	return checkEmptyResponseExpected(r, s, err)
+	}))
+	return res.noData().End()
 }
 
 // SubVolumeGroupPath returns the path to the subvolume from the root of the
@@ -90,7 +101,7 @@ func (fsa *FSAdmin) SubVolumeGroupPath(volume, name string) (string, error) {
 		"group_name": name,
 		// ceph doesn't respond in json for this cmd (even if you ask)
 	}
-	return extractPathResponse(fsa.marshalMgrCommand(m))
+	return parsePathResponse(fsa.marshalMgrCommand(m))
 }
 
 // CreateSubVolumeGroupSnapshot creates a new snapshot from the source subvolume group.
@@ -105,7 +116,7 @@ func (fsa *FSAdmin) CreateSubVolumeGroupSnapshot(volume, group, name string) err
 		"snap_name":  name,
 		"format":     "json",
 	}
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+	return fsa.marshalMgrCommand(m).noData().End()
 }
 
 // RemoveSubVolumeGroupSnapshot removes the specified snapshot from the subvolume group.
@@ -113,6 +124,18 @@ func (fsa *FSAdmin) CreateSubVolumeGroupSnapshot(volume, group, name string) err
 // Similar To:
 //  ceph fs subvolumegroup snapshot rm <volume> <group> <name>
 func (fsa *FSAdmin) RemoveSubVolumeGroupSnapshot(volume, group, name string) error {
+	return fsa.rmSubVolumeGroupSnapshot(volume, group, name, rmFlags{})
+}
+
+// ForceRemoveSubVolumeGroupSnapshot removes the specified snapshot from the subvolume group.
+//
+// Similar To:
+//  ceph fs subvolumegroup snapshot rm <volume> <group> <name> --force
+func (fsa *FSAdmin) ForceRemoveSubVolumeGroupSnapshot(volume, group, name string) error {
+	return fsa.rmSubVolumeGroupSnapshot(volume, group, name, rmFlags{force: true})
+}
+
+func (fsa *FSAdmin) rmSubVolumeGroupSnapshot(volume, group, name string, o rmFlags) error {
 	m := map[string]string{
 		"prefix":     "fs subvolumegroup snapshot rm",
 		"vol_name":   volume,
@@ -120,7 +143,7 @@ func (fsa *FSAdmin) RemoveSubVolumeGroupSnapshot(volume, group, name string) err
 		"snap_name":  name,
 		"format":     "json",
 	}
-	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+	return fsa.marshalMgrCommand(o.Update(m)).noData().End()
 }
 
 // ListSubVolumeGroupSnapshots returns a listing of snapshots for a given subvolume group.
diff --git a/cephfs/admin/volume.go b/cephfs/admin/volume.go
index f77ffa9..ea9fea5 100644
--- a/cephfs/admin/volume.go
+++ b/cephfs/admin/volume.go
@@ -12,8 +12,8 @@ var (
 // Similar To:
 //  ceph fs volume ls
 func (fsa *FSAdmin) ListVolumes() ([]string, error) {
-	r, s, err := fsa.rawMgrCommand(listVolumesCmd)
-	return parseListNames(r, s, err)
+	res := fsa.rawMgrCommand(listVolumesCmd)
+	return parseListNames(res)
 }
 
 // VolumeIdent contains a pair of file system identifying values: the volume
@@ -39,14 +39,17 @@ const (
 	dumpOkLen    = len(dumpOkPrefix)
 )
 
-func parseDumpToIdents(r []byte, s string, err error) ([]VolumeIdent, error) {
-	if len(s) >= dumpOkLen && s[:dumpOkLen] == dumpOkPrefix {
+func parseDumpToIdents(res response) ([]VolumeIdent, error) {
+	if !res.Ok() {
+		return nil, res.End()
+	}
+	if len(res.status) >= dumpOkLen && res.status[:dumpOkLen] == dumpOkPrefix {
 		// Unhelpfully, ceph drops a status string on success responses for this
 		// call. this hacks around that by ignoring its typical prefix
-		s = ""
+		res.status = ""
 	}
 	var dump fsDump
-	if err := unmarshalResponseJSON(r, s, err, &dump); err != nil {
+	if err := res.noStatus().unmarshal(&dump).End(); err != nil {
 		return nil, err
 	}
 	// copy the dump json into the simpler enumeration list
@@ -64,6 +67,5 @@ func (fsa *FSAdmin) EnumerateVolumes() ([]VolumeIdent, error) {
 	// only way to do it, but it's the only one I know of currently. Because of
 	// this and to keep our initial implementation simple, we expose our own
 	// simplified type only, rather do a partial implementation of dump.
-	r, s, err := fsa.rawMonCommand(dumpVolumesCmd)
-	return parseDumpToIdents(r, s, err)
+	return parseDumpToIdents(fsa.rawMonCommand(dumpVolumesCmd))
 }
diff --git a/cephfs/admin/volume_octopus.go b/cephfs/admin/volume_octopus.go
index b75c9ca..6ce17a1 100644
--- a/cephfs/admin/volume_octopus.go
+++ b/cephfs/admin/volume_octopus.go
@@ -18,9 +18,9 @@ type VolumeStatus struct {
 	Pools      []VolumePool `json:"pools"`
 }
 
-func parseVolumeStatus(res []byte, status string, err error) (*VolumeStatus, error) {
+func parseVolumeStatus(res response) (*VolumeStatus, error) {
 	var vs VolumeStatus
-	err = unmarshalResponseJSON(res, status, err, &vs)
+	err := res.noStatus().unmarshal(&vs).End()
 	return &vs, err
 }
 
@@ -29,10 +29,10 @@ func parseVolumeStatus(res []byte, status string, err error) (*VolumeStatus, err
 // Similar To:
 //  ceph fs status cephfs <name>
 func (fsa *FSAdmin) VolumeStatus(name string) (*VolumeStatus, error) {
-	r, s, err := fsa.marshalMgrCommand(map[string]string{
+	res := fsa.marshalMgrCommand(map[string]string{
 		"fs":     name,
 		"prefix": "fs status",
 		"format": "json",
 	})
-	return parseVolumeStatus(r, s, err)
+	return parseVolumeStatus(res)
 }


Test patch:
diff --git a/cephfs/admin/clone_test.go b/cephfs/admin/clone_test.go
new file mode 100644
index 0000000..4dc3161
--- /dev/null
+++ b/cephfs/admin/clone_test.go
@@ -0,0 +1,208 @@
+// +build !luminous,!mimic
+
+package admin
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var sampleCloneStatusPending = []byte(`{
+  "status": {
+    "state": "pending",
+    "source": {
+      "volume": "cephfs",
+      "subvolume": "jurrasic",
+      "snapshot": "dinodna",
+      "group": "park"
+    }
+  } 
+}`)
+
+var sampleCloneStatusInProg = []byte(`{
+  "status": {
+    "state": "in-progress",
+    "source": {
+      "volume": "cephfs",
+      "subvolume": "subvol1",
+      "snapshot": "snap1"
+    }
+  }
+}`)
+
+func TestParseCloneStatus(t *testing.T) {
+	R := newResponse
+	t.Run("error", func(t *testing.T) {
+		_, err := parseCloneStatus(R(nil, "", errors.New("flub")))
+		assert.Error(t, err)
+		assert.Equal(t, "flub", err.Error())
+	})
+	t.Run("statusSet", func(t *testing.T) {
+		_, err := parseCloneStatus(R(nil, "unexpected!", nil))
+		assert.Error(t, err)
+	})
+	t.Run("badJSON", func(t *testing.T) {
+		_, err := parseCloneStatus(R([]byte("_XxXxX"), "", nil))
+		assert.Error(t, err)
+	})
+	t.Run("okPending", func(t *testing.T) {
+		status, err := parseCloneStatus(R(sampleCloneStatusPending, "", nil))
+		assert.NoError(t, err)
+		if assert.NotNil(t, status) {
+			assert.EqualValues(t, ClonePending, status.State)
+			assert.EqualValues(t, "cephfs", status.Source.Volume)
+			assert.EqualValues(t, "jurrasic", status.Source.SubVolume)
+			assert.EqualValues(t, "dinodna", status.Source.Snapshot)
+			assert.EqualValues(t, "park", status.Source.Group)
+		}
+	})
+	t.Run("okInProg", func(t *testing.T) {
+		status, err := parseCloneStatus(R(sampleCloneStatusInProg, "", nil))
+		assert.NoError(t, err)
+		if assert.NotNil(t, status) {
+			assert.EqualValues(t, CloneInProgress, status.State)
+			assert.EqualValues(t, "cephfs", status.Source.Volume)
+			assert.EqualValues(t, "subvol1", status.Source.SubVolume)
+			assert.EqualValues(t, "snap1", status.Source.Snapshot)
+			assert.EqualValues(t, "", status.Source.Group)
+		}
+	})
+}
+
+func TestCloneSubVolumeSnapshot(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "Park"
+	subname := "Jurrasic"
+	snapname := "dinodna0"
+	clonename := "babydino"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolumeGroup(volume, group)
+		assert.NoError(t, err)
+	}()
+
+	svopts := &SubVolumeOptions{
+		Mode: 0750,
+		Size: 20 * gibiByte,
+	}
+	err = fsa.CreateSubVolume(volume, group, subname, svopts)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolume(volume, group, subname)
+		assert.NoError(t, err)
+	}()
+
+	err = fsa.CreateSubVolumeSnapshot(volume, group, subname, snapname)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolumeSnapshot(volume, group, subname, snapname)
+		assert.NoError(t, err)
+	}()
+
+	err = fsa.CloneSubVolumeSnapshot(
+		volume, group, subname, snapname, clonename,
+		&CloneOptions{TargetGroup: group})
+	var x NotProtectedError
+	if errors.As(err, &x) {
+		err = fsa.ProtectSubVolumeSnapshot(volume, group, subname, snapname)
+		assert.NoError(t, err)
+		defer func() {
+			err := fsa.UnprotectSubVolumeSnapshot(volume, group, subname, snapname)
+			assert.NoError(t, err)
+		}()
+
+		err = fsa.CloneSubVolumeSnapshot(
+			volume, group, subname, snapname, clonename,
+			&CloneOptions{TargetGroup: group})
+	}
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolume(volume, group, clonename)
+		assert.NoError(t, err)
+	}()
+
+	for done := false; !done; {
+		status, err := fsa.CloneStatus(volume, group, clonename)
+		assert.NoError(t, err)
+		assert.NotNil(t, status)
+		switch status.State {
+		case ClonePending, CloneInProgress:
+		case CloneComplete:
+			done = true
+		case CloneFailed:
+			t.Fatal("clone failed")
+		default:
+			t.Fatalf("invalid status.State: %q", status.State)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func TestCancelClone(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "Park"
+	subname := "Jurrasic"
+	snapname := "dinodna0"
+	clonename := "babydino2"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolumeGroup(volume, group)
+		assert.NoError(t, err)
+	}()
+
+	svopts := &SubVolumeOptions{
+		Mode: 0750,
+		Size: 20 * gibiByte,
+	}
+	err = fsa.CreateSubVolume(volume, group, subname, svopts)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolume(volume, group, subname)
+		assert.NoError(t, err)
+	}()
+
+	err = fsa.CreateSubVolumeSnapshot(volume, group, subname, snapname)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolumeSnapshot(volume, group, subname, snapname)
+		assert.NoError(t, err)
+	}()
+
+	err = fsa.CloneSubVolumeSnapshot(
+		volume, group, subname, snapname, clonename,
+		&CloneOptions{TargetGroup: group})
+	var x NotProtectedError
+	if errors.As(err, &x) {
+		err = fsa.ProtectSubVolumeSnapshot(volume, group, subname, snapname)
+		assert.NoError(t, err)
+		defer func() {
+			err := fsa.UnprotectSubVolumeSnapshot(volume, group, subname, snapname)
+			assert.NoError(t, err)
+		}()
+
+		err = fsa.CloneSubVolumeSnapshot(
+			volume, group, subname, snapname, clonename,
+			&CloneOptions{TargetGroup: group})
+	}
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.ForceRemoveSubVolume(volume, group, clonename)
+		assert.NoError(t, err)
+	}()
+
+	// we can't guarantee that this clone is can be canceled here, especially
+	// if the clone happens fast on the ceph server side, but I have not yet
+	// seen an instance where it fails. If it happens we can adjust the test as
+	// needed.
+	err = fsa.CancelClone(volume, group, clonename)
+	assert.NoError(t, err)
+}
diff --git a/cephfs/admin/fsadmin_test.go b/cephfs/admin/fsadmin_test.go
index 05dd11b..327b695 100644
--- a/cephfs/admin/fsadmin_test.go
+++ b/cephfs/admin/fsadmin_test.go
@@ -93,8 +93,8 @@ func getFSAdmin(t *testing.T) *FSAdmin {
 
 func TestInvalidFSAdmin(t *testing.T) {
 	fsa := &FSAdmin{}
-	_, _, err := fsa.rawMgrCommand([]byte("FOOBAR!"))
-	assert.Error(t, err)
+	res := fsa.rawMgrCommand([]byte("FOOBAR!"))
+	assert.Error(t, res.Unwrap())
 }
 
 type badMarshalType bool
@@ -107,26 +107,27 @@ func TestBadMarshal(t *testing.T) {
 	fsa := getFSAdmin(t)
 
 	var bad badMarshalType
-	_, _, err := fsa.marshalMgrCommand(bad)
-	assert.Error(t, err)
+	res := fsa.marshalMgrCommand(bad)
+	assert.Error(t, res.Unwrap())
 }
 
 func TestParseListNames(t *testing.T) {
+	R := newResponse
 	t.Run("error", func(t *testing.T) {
-		_, err := parseListNames(nil, "", errors.New("bonk"))
+		_, err := parseListNames(R(nil, "", errors.New("bonk")))
 		assert.Error(t, err)
 		assert.Equal(t, "bonk", err.Error())
 	})
 	t.Run("statusSet", func(t *testing.T) {
-		_, err := parseListNames(nil, "unexpected!", nil)
+		_, err := parseListNames(R(nil, "unexpected!", nil))
 		assert.Error(t, err)
 	})
 	t.Run("badJSON", func(t *testing.T) {
-		_, err := parseListNames([]byte("Foo[[["), "", nil)
+		_, err := parseListNames(R([]byte("Foo[[["), "", nil))
 		assert.Error(t, err)
 	})
 	t.Run("ok", func(t *testing.T) {
-		l, err := parseListNames([]byte(`[{"name":"bob"}]`), "", nil)
+		l, err := parseListNames(R([]byte(`[{"name":"bob"}]`), "", nil))
 		assert.NoError(t, err)
 		if assert.Len(t, l, 1) {
 			assert.Equal(t, "bob", l[0])
@@ -135,21 +136,22 @@ func TestParseListNames(t *testing.T) {
 }
 
 func TestCheckEmptyResponseExpected(t *testing.T) {
+	R := newResponse
 	t.Run("error", func(t *testing.T) {
-		err := checkEmptyResponseExpected(nil, "", errors.New("bonk"))
+		err := R(nil, "", errors.New("bonk")).noData().End()
 		assert.Error(t, err)
 		assert.Equal(t, "bonk", err.Error())
 	})
 	t.Run("statusSet", func(t *testing.T) {
-		err := checkEmptyResponseExpected(nil, "unexpected!", nil)
+		err := R(nil, "unexpected!", nil).noData().End()
 		assert.Error(t, err)
 	})
 	t.Run("someJSON", func(t *testing.T) {
-		err := checkEmptyResponseExpected([]byte(`{"trouble": true}`), "", nil)
+		err := R([]byte(`{"trouble": true}`), "", nil).noData().End()
 		assert.Error(t, err)
 	})
 	t.Run("ok", func(t *testing.T) {
-		err := checkEmptyResponseExpected([]byte{}, "", nil)
+		err := R([]byte{}, "", nil).noData().End()
 		assert.NoError(t, err)
 	})
 }
diff --git a/cephfs/admin/response_test.go b/cephfs/admin/response_test.go
new file mode 100644
index 0000000..1633ad7
--- /dev/null
+++ b/cephfs/admin/response_test.go
@@ -0,0 +1,128 @@
+// +build !luminous,!mimic
+
+package admin
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResponse(t *testing.T) {
+	e1 := errors.New("error one")
+	e2 := errors.New("error two")
+	r1 := response{
+		body: []byte(`{"foo": "bar", "baz": 1}`),
+	}
+	r2 := response{
+		status: "System notice: disabled for maintenance",
+		err:    e1,
+	}
+	r3 := response{
+		body:   []byte(`{"oof": "RAB", "baz": 8}`),
+		status: "reversed polarity detected",
+	}
+	r4 := response{
+		body:   []byte(`{"whoops": true, "state": "total protonic reversal"}`),
+		status: "",
+		err:    e2,
+	}
+
+	t.Run("ok", func(t *testing.T) {
+		assert.True(t, r1.Ok())
+		assert.False(t, r2.Ok())
+		assert.True(t, r3.Ok())
+	})
+
+	t.Run("error", func(t *testing.T) {
+		assert.Equal(t,
+			"error one: \"System notice: disabled for maintenance\"",
+			r2.Error())
+		assert.Equal(t,
+			e2.Error(),
+			r4.Error())
+	})
+
+	t.Run("unwrap", func(t *testing.T) {
+		assert.Equal(t, e1, r2.Unwrap())
+		assert.Equal(t, e2, r4.Unwrap())
+	})
+
+	t.Run("status", func(t *testing.T) {
+		assert.Equal(t, "", r1.Status())
+		assert.Equal(t, "System notice: disabled for maintenance", r2.Status())
+		assert.Equal(t, "reversed polarity detected", r3.Status())
+	})
+
+	t.Run("end", func(t *testing.T) {
+		assert.Nil(t, r1.End())
+		assert.NotNil(t, r2.End())
+		assert.EqualValues(t, r2, r2.End())
+	})
+
+	t.Run("noStatus", func(t *testing.T) {
+		assert.EqualValues(t, r1, r1.noStatus())
+		assert.EqualValues(t, r2, r2.noStatus())
+
+		x := r3.noStatus()
+		assert.EqualValues(t, ErrStatusNotEmpty, x.Unwrap())
+		assert.EqualValues(t, r3.Status(), x.Status())
+	})
+
+	t.Run("noBody", func(t *testing.T) {
+		x := r1.noBody()
+		assert.EqualValues(t, ErrBodyNotEmpty, x.Unwrap())
+		assert.EqualValues(t, r1.Status(), x.Status())
+
+		assert.EqualValues(t, r2, r2.noBody())
+
+		rtemp := response{}
+		assert.EqualValues(t, rtemp, rtemp.noBody())
+	})
+
+	t.Run("noData", func(t *testing.T) {
+		x := r1.noData()
+		assert.EqualValues(t, ErrBodyNotEmpty, x.Unwrap())
+		assert.EqualValues(t, r1.Status(), x.Status())
+
+		x = r3.noStatus()
+		assert.EqualValues(t, ErrStatusNotEmpty, x.Unwrap())
+		assert.EqualValues(t, r3.Status(), x.Status())
+
+		rtemp := response{}
+		assert.EqualValues(t, rtemp, rtemp.noData())
+	})
+
+	t.Run("filterDeprecated", func(t *testing.T) {
+		assert.EqualValues(t, r1, r1.filterDeprecated())
+		assert.EqualValues(t, r2, r2.filterDeprecated())
+
+		rtemp := response{
+			status: "blorple call is deprecated and will be removed in a future release",
+		}
+		x := rtemp.filterDeprecated()
+		assert.True(t, x.Ok())
+		assert.Nil(t, x.End())
+		assert.Equal(t, "", x.Status())
+	})
+
+	t.Run("unmarshal", func(t *testing.T) {
+		var v map[string]interface{}
+		assert.EqualValues(t, r1, r1.unmarshal(&v))
+		assert.EqualValues(t, "bar", v["foo"])
+
+		assert.EqualValues(t, r2, r2.unmarshal(&v))
+
+		rtemp := response{body: []byte("foo!")}
+		x := rtemp.unmarshal(&v)
+		assert.False(t, x.Ok())
+		assert.Contains(t, x.Error(), "invalid character")
+	})
+
+	t.Run("newResponse", func(t *testing.T) {
+		rtemp := newResponse(nil, "x", e2)
+		assert.False(t, rtemp.Ok())
+		assert.Equal(t, "x", rtemp.Status())
+	})
+}
diff --git a/cephfs/admin/subvolume_octopus_test.go b/cephfs/admin/subvolume_octopus_test.go
index 163cbbb..836473e 100644
--- a/cephfs/admin/subvolume_octopus_test.go
+++ b/cephfs/admin/subvolume_octopus_test.go
@@ -20,21 +20,22 @@ var sampleSubVolumeSnapshoInfo1 = []byte(`
 `)
 
 func TestParseSubVolumeSnapshotInfo(t *testing.T) {
+	R := newResponse
 	t.Run("error", func(t *testing.T) {
-		_, err := parseSubVolumeSnapshotInfo(nil, "", errors.New("flub"))
+		_, err := parseSubVolumeSnapshotInfo(R(nil, "", errors.New("flub")))
 		assert.Error(t, err)
 		assert.Equal(t, "flub", err.Error())
 	})
 	t.Run("statusSet", func(t *testing.T) {
-		_, err := parseSubVolumeSnapshotInfo(nil, "unexpected!", nil)
+		_, err := parseSubVolumeSnapshotInfo(R(nil, "unexpected!", nil))
 		assert.Error(t, err)
 	})
 	t.Run("badJSON", func(t *testing.T) {
-		_, err := parseSubVolumeSnapshotInfo([]byte("_XxXxX"), "", nil)
+		_, err := parseSubVolumeSnapshotInfo(R([]byte("_XxXxX"), "", nil))
 		assert.Error(t, err)
 	})
 	t.Run("ok", func(t *testing.T) {
-		info, err := parseSubVolumeSnapshotInfo(sampleSubVolumeSnapshoInfo1, "", nil)
+		info, err := parseSubVolumeSnapshotInfo(R(sampleSubVolumeSnapshoInfo1, "", nil))
 		assert.NoError(t, err)
 		if assert.NotNil(t, info) {
 			assert.Equal(t, "cephfs_data", info.DataPool)
diff --git a/cephfs/admin/subvolume_test.go b/cephfs/admin/subvolume_test.go
index 5a54ccf..84d489e 100644
--- a/cephfs/admin/subvolume_test.go
+++ b/cephfs/admin/subvolume_test.go
@@ -113,25 +113,34 @@ func TestRemoveSubVolume(t *testing.T) {
 	assert.NoError(t, err)
 	beforeCount := len(lsv)
 
-	err = fsa.CreateSubVolume(volume, NoGroup, "deletemev1", nil)
-	assert.NoError(t, err)
+	removeTest := func(t *testing.T, rm func(string, string, string) error) {
+		err = fsa.CreateSubVolume(volume, NoGroup, "deletemev1", nil)
+		assert.NoError(t, err)
 
-	lsv, err = fsa.ListSubVolumes(volume, NoGroup)
-	assert.NoError(t, err)
-	afterCount := len(lsv)
-	assert.Equal(t, beforeCount, afterCount-1)
+		lsv, err = fsa.ListSubVolumes(volume, NoGroup)
+		assert.NoError(t, err)
+		afterCount := len(lsv)
+		assert.Equal(t, beforeCount, afterCount-1)
 
-	err = fsa.RemoveSubVolume(volume, NoGroup, "deletemev1")
-	assert.NoError(t, err)
+		err = rm(volume, NoGroup, "deletemev1")
+		assert.NoError(t, err)
 
-	delay()
-	lsv, err = fsa.ListSubVolumes(volume, NoGroup)
-	assert.NoError(t, err)
-	nowCount := len(lsv)
-	if !assert.Equal(t, beforeCount, nowCount) {
-		// this is a hack for debugging a flapping test
-		assert.Equal(t, []string{}, lsv)
+		delay()
+		lsv, err = fsa.ListSubVolumes(volume, NoGroup)
+		assert.NoError(t, err)
+		nowCount := len(lsv)
+		if !assert.Equal(t, beforeCount, nowCount) {
+			// this is a hack for debugging a flapping test
+			assert.Equal(t, []string{}, lsv)
+		}
 	}
+
+	t.Run("standard", func(t *testing.T) {
+		removeTest(t, fsa.RemoveSubVolume)
+	})
+	t.Run("force", func(t *testing.T) {
+		removeTest(t, fsa.ForceRemoveSubVolume)
+	})
 }
 
 func TestResizeSubVolume(t *testing.T) {
@@ -266,17 +275,18 @@ var badSampleSubVolumeInfo2 = []byte(`
 `)
 
 func TestParseSubVolumeInfo(t *testing.T) {
+	R := newResponse
 	t.Run("error", func(t *testing.T) {
-		_, err := parseSubVolumeInfo(nil, "", errors.New("gleep glop"))
+		_, err := parseSubVolumeInfo(R(nil, "", errors.New("gleep glop")))
 		assert.Error(t, err)
 		assert.Equal(t, "gleep glop", err.Error())
 	})
 	t.Run("statusSet", func(t *testing.T) {
-		_, err := parseSubVolumeInfo(nil, "unexpected!", nil)
+		_, err := parseSubVolumeInfo(R(nil, "unexpected!", nil))
 		assert.Error(t, err)
 	})
 	t.Run("ok", func(t *testing.T) {
-		info, err := parseSubVolumeInfo(sampleSubVolumeInfo1, "", nil)
+		info, err := parseSubVolumeInfo(R(sampleSubVolumeInfo1, "", nil))
 		assert.NoError(t, err)
 		if assert.NotNil(t, info) {
 			assert.Equal(t,
@@ -290,7 +300,7 @@ func TestParseSubVolumeInfo(t *testing.T) {
 		}
 	})
 	t.Run("ok2", func(t *testing.T) {
-		info, err := parseSubVolumeInfo(sampleSubVolumeInfo2, "", nil)
+		info, err := parseSubVolumeInfo(R(sampleSubVolumeInfo2, "", nil))
 		assert.NoError(t, err)
 		if assert.NotNil(t, info) {
 			assert.Equal(t,
@@ -304,12 +314,12 @@ func TestParseSubVolumeInfo(t *testing.T) {
 		}
 	})
 	t.Run("invalidBytesQuotaValue", func(t *testing.T) {
-		info, err := parseSubVolumeInfo(badSampleSubVolumeInfo1, "", nil)
+		info, err := parseSubVolumeInfo(R(badSampleSubVolumeInfo1, "", nil))
 		assert.Error(t, err)
 		assert.Nil(t, info)
 	})
 	t.Run("invalidBytesQuotaType", func(t *testing.T) {
-		info, err := parseSubVolumeInfo(badSampleSubVolumeInfo2, "", nil)
+		info, err := parseSubVolumeInfo(R(badSampleSubVolumeInfo2, "", nil))
 		assert.Error(t, err)
 		assert.Nil(t, info)
 	})
@@ -381,6 +391,13 @@ func TestSubVolumeSnapshots(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	t.Run("createAndForceRemove", func(t *testing.T) {
+		err = fsa.CreateSubVolumeSnapshot(volume, group, subname, snapname1)
+		assert.NoError(t, err)
+		err := fsa.ForceRemoveSubVolumeSnapshot(volume, group, subname, snapname1)
+		assert.NoError(t, err)
+	})
+
 	t.Run("listOne", func(t *testing.T) {
 		err = fsa.CreateSubVolumeSnapshot(volume, group, subname, snapname1)
 		assert.NoError(t, err)
diff --git a/cephfs/admin/subvolumegroup_test.go b/cephfs/admin/subvolumegroup_test.go
index b57e99c..04da168 100644
--- a/cephfs/admin/subvolumegroup_test.go
+++ b/cephfs/admin/subvolumegroup_test.go
@@ -71,21 +71,30 @@ func TestRemoveSubVolumeGroup(t *testing.T) {
 	assert.NoError(t, err)
 	beforeCount := len(lsvg)
 
-	err = fsa.CreateSubVolumeGroup(volume, "deleteme1", nil)
-	assert.NoError(t, err)
+	removeTest := func(t *testing.T, rm func(string, string) error) {
+		err = fsa.CreateSubVolumeGroup(volume, "deleteme1", nil)
+		assert.NoError(t, err)
 
-	lsvg, err = fsa.ListSubVolumeGroups(volume)
-	assert.NoError(t, err)
-	afterCount := len(lsvg)
-	assert.Equal(t, beforeCount, afterCount-1)
+		lsvg, err = fsa.ListSubVolumeGroups(volume)
+		assert.NoError(t, err)
+		afterCount := len(lsvg)
+		assert.Equal(t, beforeCount, afterCount-1)
 
-	err = fsa.RemoveSubVolumeGroup(volume, "deleteme1")
-	assert.NoError(t, err)
+		err = rm(volume, "deleteme1")
+		assert.NoError(t, err)
 
-	lsvg, err = fsa.ListSubVolumeGroups(volume)
-	assert.NoError(t, err)
-	nowCount := len(lsvg)
-	assert.Equal(t, beforeCount, nowCount)
+		lsvg, err = fsa.ListSubVolumeGroups(volume)
+		assert.NoError(t, err)
+		nowCount := len(lsvg)
+		assert.Equal(t, beforeCount, nowCount)
+	}
+
+	t.Run("standard", func(t *testing.T) {
+		removeTest(t, fsa.RemoveSubVolumeGroup)
+	})
+	t.Run("force", func(t *testing.T) {
+		removeTest(t, fsa.ForceRemoveSubVolumeGroup)
+	})
 }
 
 func TestSubVolumeGroupPath(t *testing.T) {
@@ -144,6 +153,13 @@ func TestSubVolumeGroupSnapshots(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	t.Run("createAndForceRemove", func(t *testing.T) {
+		err = fsa.CreateSubVolumeGroupSnapshot(volume, group, snapname1)
+		assert.NoError(t, err)
+		err := fsa.ForceRemoveSubVolumeGroupSnapshot(volume, group, snapname1)
+		assert.NoError(t, err)
+	})
+
 	t.Run("listOne", func(t *testing.T) {
 		err = fsa.CreateSubVolumeGroupSnapshot(volume, group, snapname1)
 		assert.NoError(t, err)
diff --git a/cephfs/admin/volume_octopus_test.go b/cephfs/admin/volume_octopus_test.go
index bf8b87b..cd8fc85 100644
--- a/cephfs/admin/volume_octopus_test.go
+++ b/cephfs/admin/volume_octopus_test.go
@@ -27,21 +27,22 @@ var sampleVolumeStatus1 = []byte(`
 `)
 
 func TestParseVolumeStatus(t *testing.T) {
+	R := newResponse
 	t.Run("error", func(t *testing.T) {
-		_, err := parseVolumeStatus(nil, "", errors.New("bonk"))
+		_, err := parseVolumeStatus(R(nil, "", errors.New("bonk")))
 		assert.Error(t, err)
 		assert.Equal(t, "bonk", err.Error())
 	})
 	t.Run("statusSet", func(t *testing.T) {
-		_, err := parseVolumeStatus(nil, "unexpected!", nil)
+		_, err := parseVolumeStatus(R(nil, "unexpected!", nil))
 		assert.Error(t, err)
 	})
 	t.Run("badJSON", func(t *testing.T) {
-		_, err := parseVolumeStatus([]byte("_XxXxX"), "", nil)
+		_, err := parseVolumeStatus(R([]byte("_XxXxX"), "", nil))
 		assert.Error(t, err)
 	})
 	t.Run("ok", func(t *testing.T) {
-		s, err := parseVolumeStatus(sampleVolumeStatus1, "", nil)
+		s, err := parseVolumeStatus(R(sampleVolumeStatus1, "", nil))
 		assert.NoError(t, err)
 		if assert.NotNil(t, s) {
 			assert.Contains(t, s.MDSVersion, "ceph version 15.2.4")
diff --git a/cephfs/admin/volume_test.go b/cephfs/admin/volume_test.go
index 770acb4..e909e5a 100644
--- a/cephfs/admin/volume_test.go
+++ b/cephfs/admin/volume_test.go
@@ -140,19 +140,20 @@ var sampleDump2 = []byte(`
 `)
 
 func TestParseDumpToIdents(t *testing.T) {
+	R := newResponse
 	fakePrefix := dumpOkPrefix + " 5"
 	t.Run("error", func(t *testing.T) {
-		idents, err := parseDumpToIdents(nil, "", errors.New("boop"))
+		idents, err := parseDumpToIdents(R(nil, "", errors.New("boop")))
 		assert.Error(t, err)
 		assert.Equal(t, "boop", err.Error())
 		assert.Nil(t, idents)
 	})
 	t.Run("badStatus", func(t *testing.T) {
-		_, err := parseDumpToIdents(sampleDump1, "unexpected!", nil)
+		_, err := parseDumpToIdents(R(sampleDump1, "unexpected!", nil))
 		assert.Error(t, err)
 	})
 	t.Run("oneVolOk", func(t *testing.T) {
-		idents, err := parseDumpToIdents(sampleDump1, fakePrefix, nil)
+		idents, err := parseDumpToIdents(R(sampleDump1, fakePrefix, nil))
 		assert.NoError(t, err)
 		if assert.Len(t, idents, 1) {
 			assert.Equal(t, "cephfs", idents[0].Name)
@@ -160,7 +161,7 @@ func TestParseDumpToIdents(t *testing.T) {
 		}
 	})
 	t.Run("twoVolOk", func(t *testing.T) {
-		idents, err := parseDumpToIdents(sampleDump2, fakePrefix, nil)
+		idents, err := parseDumpToIdents(R(sampleDump2, fakePrefix, nil))
 		assert.NoError(t, err)
 		if assert.Len(t, idents, 2) {
 			assert.Equal(t, "wiffleball", idents[0].Name)
@@ -170,7 +171,7 @@ func TestParseDumpToIdents(t *testing.T) {
 		}
 	})
 	t.Run("unexpectedStatus", func(t *testing.T) {
-		idents, err := parseDumpToIdents(sampleDump1, "slip-up", nil)
+		idents, err := parseDumpToIdents(R(sampleDump1, "slip-up", nil))
 		assert.Error(t, err)
 		assert.Nil(t, idents)
 	})