Repository: ceph/go-ceph
Pull request #373 (2020-09-30 20:28:29)

cephfs admin: structural clean-ups
These changes depend on those in PR #372. Please review that first.

Previously, we were only building some functions based on the version of ceph client libs, but the functions are tied to the *server* version. Make the functions built and tested all the time with conditional handling for the server where needed.

## Checklist
- [x] Added tests for features and functional changes
- [x] Public functions and types are documented
- [x] Standard formatting is applied to Go code



Patch:
diff --git a/cephfs/admin/response.go b/cephfs/admin/response.go
index fb0e407..a1b6bf3 100644
--- a/cephfs/admin/response.go
+++ b/cephfs/admin/response.go
@@ -20,8 +20,26 @@ var (
 
 const (
 	deprecatedSuffix = "call is deprecated and will be removed in a future release"
+	missingPrefix    = "No handler found"
+	einval           = -22
 )
 
+type cephError interface {
+	ErrorCode() int
+}
+
+// NotImplementedError error values will be returned in the case that an API
+// call is not available in the version of Ceph that is running in the target
+// cluster.
+type NotImplementedError struct {
+	response
+}
+
+// Error implements the error interface.
+func (e NotImplementedError) Error() string {
+	return fmt.Sprintf("API call not implemented server-side: %s", e.status)
+}
+
 // response encapsulates the data returned by ceph and supports easy processing
 // pipelines.
 type response struct {
@@ -57,6 +75,11 @@ func (r response) Status() string {
 // that response is no longer needed for processing.
 func (r response) End() error {
 	if !r.Ok() {
+		if ce, ok := r.err.(cephError); ok {
+			if ce.ErrorCode() == einval && strings.HasPrefix(r.status, missingPrefix) {
+				return NotImplementedError{response: r}
+			}
+		}
 		return r
 	}
 	return nil
diff --git a/cephfs/admin/subvolume.go b/cephfs/admin/subvolume.go
index 6a0c2c2..106814f 100644
--- a/cephfs/admin/subvolume.go
+++ b/cephfs/admin/subvolume.go
@@ -291,6 +291,41 @@ func (fsa *FSAdmin) ListSubVolumeSnapshots(volume, group, name string) ([]string
 	return parseListNames(fsa.marshalMgrCommand(m))
 }
 
+// SubVolumeSnapshotInfo reports various informational values about a subvolume.
+type SubVolumeSnapshotInfo struct {
+	CreatedAt        TimeStamp `json:"created_at"`
+	DataPool         string    `json:"data_pool"`
+	HasPendingClones string    `json:"has_pending_clones"`
+	Protected        string    `json:"protected"`
+	Size             ByteCount `json:"size"`
+}
+
+func parseSubVolumeSnapshotInfo(res response) (*SubVolumeSnapshotInfo, error) {
+	var info SubVolumeSnapshotInfo
+	if err := res.noStatus().unmarshal(&info).End(); err != nil {
+		return nil, err
+	}
+	return &info, nil
+}
+
+// SubVolumeSnapshotInfo returns information about the specified subvolume snapshot.
+//
+// Similar To:
+//  ceph fs subvolume snapshot info <volume> --group-name=<group> <subvolume> <name>
+func (fsa *FSAdmin) SubVolumeSnapshotInfo(volume, group, subvolume, name string) (*SubVolumeSnapshotInfo, error) {
+	m := map[string]string{
+		"prefix":    "fs subvolume snapshot info",
+		"vol_name":  volume,
+		"sub_name":  subvolume,
+		"snap_name": name,
+		"format":    "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return parseSubVolumeSnapshotInfo(fsa.marshalMgrCommand(m))
+}
+
 // ProtectSubVolumeSnapshot protects the specified snapshot.
 //
 // Similar To:
diff --git a/cephfs/admin/subvolume_octopus.go b/cephfs/admin/subvolume_octopus.go
deleted file mode 100644
index afb6ca7..0000000
--- a/cephfs/admin/subvolume_octopus.go
+++ /dev/null
@@ -1,38 +0,0 @@
-// +build octopus
-
-package admin
-
-// SubVolumeSnapshotInfo reports various informational values about a subvolume.
-type SubVolumeSnapshotInfo struct {
-	CreatedAt        TimeStamp `json:"created_at"`
-	DataPool         string    `json:"data_pool"`
-	HasPendingClones string    `json:"has_pending_clones"`
-	Protected        string    `json:"protected"`
-	Size             ByteCount `json:"size"`
-}
-
-func parseSubVolumeSnapshotInfo(res response) (*SubVolumeSnapshotInfo, error) {
-	var info SubVolumeSnapshotInfo
-	if err := res.noStatus().unmarshal(&info).End(); err != nil {
-		return nil, err
-	}
-	return &info, nil
-}
-
-// SubVolumeSnapshotInfo returns information about the specified subvolume snapshot.
-//
-// Similar To:
-//  ceph fs subvolume snapshot info <volume> --group-name=<group> <subvolume> <name>
-func (fsa *FSAdmin) SubVolumeSnapshotInfo(volume, group, subvolume, name string) (*SubVolumeSnapshotInfo, error) {
-	m := map[string]string{
-		"prefix":    "fs subvolume snapshot info",
-		"vol_name":  volume,
-		"sub_name":  subvolume,
-		"snap_name": name,
-		"format":    "json",
-	}
-	if group != NoGroup {
-		m["group_name"] = group
-	}
-	return parseSubVolumeSnapshotInfo(fsa.marshalMgrCommand(m))
-}
diff --git a/cephfs/admin/volume.go b/cephfs/admin/volume.go
index ea9fea5..0e7ccf5 100644
--- a/cephfs/admin/volume.go
+++ b/cephfs/admin/volume.go
@@ -2,6 +2,10 @@
 
 package admin
 
+import (
+	"bytes"
+)
+
 var (
 	listVolumesCmd = []byte(`{"prefix":"fs volume ls"}`)
 	dumpVolumesCmd = []byte(`{"prefix":"fs dump","format":"json"}`)
@@ -37,6 +41,8 @@ type fsDump struct {
 const (
 	dumpOkPrefix = "dumped fsmap epoch"
 	dumpOkLen    = len(dumpOkPrefix)
+
+	invalidTextualResponse = "this ceph version returns a non-parsable volume status response"
 )
 
 func parseDumpToIdents(res response) ([]VolumeIdent, error) {
@@ -69,3 +75,49 @@ func (fsa *FSAdmin) EnumerateVolumes() ([]VolumeIdent, error) {
 	// simplified type only, rather do a partial implementation of dump.
 	return parseDumpToIdents(fsa.rawMonCommand(dumpVolumesCmd))
 }
+
+// VolumePool reports on the pool status for a CephFS volume.
+type VolumePool struct {
+	ID        int    `json:"id"`
+	Name      string `json:"name"`
+	Type      string `json:"type"`
+	Available uint64 `json:"avail"`
+	Used      uint64 `json:"used"`
+}
+
+// VolumeStatus reports various properties of a CephFS volume.
+// TODO: Fill in.
+type VolumeStatus struct {
+	MDSVersion string       `json:"mds_version"`
+	Pools      []VolumePool `json:"pools"`
+}
+
+func parseVolumeStatus(res response) (*VolumeStatus, error) {
+	var vs VolumeStatus
+	res = res.noStatus()
+	if !res.Ok() {
+		return nil, res.End()
+	}
+	res = res.unmarshal(&vs)
+	if !res.Ok() {
+		if bytes.HasPrefix(res.body, []byte("ceph")) {
+			res.status = invalidTextualResponse
+			return nil, NotImplementedError{response: res}
+		}
+		return nil, res.End()
+	}
+	return &vs, nil
+}
+
+// VolumeStatus returns a VolumeStatus object for the given volume name.
+//
+// Similar To:
+//  ceph fs status cephfs <name>
+func (fsa *FSAdmin) VolumeStatus(name string) (*VolumeStatus, error) {
+	res := fsa.marshalMgrCommand(map[string]string{
+		"fs":     name,
+		"prefix": "fs status",
+		"format": "json",
+	})
+	return parseVolumeStatus(res)
+}
diff --git a/cephfs/admin/volume_octopus.go b/cephfs/admin/volume_octopus.go
deleted file mode 100644
index 6ce17a1..0000000
--- a/cephfs/admin/volume_octopus.go
+++ /dev/null
@@ -1,38 +0,0 @@
-// +build octopus
-
-package admin
-
-// VolumePool reports on the pool status for a CephFS volume.
-type VolumePool struct {
-	ID        int    `json:"id"`
-	Name      string `json:"name"`
-	Type      string `json:"type"`
-	Available uint64 `json:"avail"`
-	Used      uint64 `json:"used"`
-}
-
-// VolumeStatus reports various properties of a CephFS volume.
-// TODO: Fill in.
-type VolumeStatus struct {
-	MDSVersion string       `json:"mds_version"`
-	Pools      []VolumePool `json:"pools"`
-}
-
-func parseVolumeStatus(res response) (*VolumeStatus, error) {
-	var vs VolumeStatus
-	err := res.noStatus().unmarshal(&vs).End()
-	return &vs, err
-}
-
-// VolumeStatus returns a VolumeStatus object for the given volume name.
-//
-// Similar To:
-//  ceph fs status cephfs <name>
-func (fsa *FSAdmin) VolumeStatus(name string) (*VolumeStatus, error) {
-	res := fsa.marshalMgrCommand(map[string]string{
-		"fs":     name,
-		"prefix": "fs status",
-		"format": "json",
-	})
-	return parseVolumeStatus(res)
-}


Test patch:
diff --git a/cephfs/admin/fsadmin_test.go b/cephfs/admin/fsadmin_test.go
index 327b695..dfc1294 100644
--- a/cephfs/admin/fsadmin_test.go
+++ b/cephfs/admin/fsadmin_test.go
@@ -19,6 +19,10 @@ var (
 
 	// set debugTrace to true to use tracing in tests
 	debugTrace = false
+
+	// some tests are sensitive to the server version
+	serverIsNautilus = false
+	serverIsOctopus  = false
 )
 
 func init() {
@@ -26,6 +30,27 @@ func init() {
 	if ok, err := strconv.ParseBool(dt); ok && err == nil {
 		debugTrace = true
 	}
+	switch os.Getenv("CEPH_VERSION") {
+	case "nautilus":
+		serverIsNautilus = true
+	case "octopus":
+		serverIsOctopus = true
+	}
+}
+
+func TestServerSentinel(t *testing.T) {
+	// there probably *is* a better way to do this but I'm doing what's easy
+	// and expedient at the moment. That's tying the tests to the environment
+	// var to tell us what version of the *server* we are testing against. The
+	// build tags control what version of the *client libs* we use.  These
+	// happen to be the same for our CI tests today, but its a lousy way to
+	// organize things IMO.
+	// This check is intended to fail the test suite if you don't tell it a
+	// server version it expects and force us to update the tests if a new
+	// version of ceph is added.
+	if !serverIsNautilus && !serverIsOctopus {
+		t.Fatalf("server must be nautilus or octopus (do the tests need updating?)")
+	}
 }
 
 // tracingCommander serves two purposes: first, it allows one to trace the
diff --git a/cephfs/admin/response_test.go b/cephfs/admin/response_test.go
index 1633ad7..ee00541 100644
--- a/cephfs/admin/response_test.go
+++ b/cephfs/admin/response_test.go
@@ -125,4 +125,28 @@ func TestResponse(t *testing.T) {
 		assert.False(t, rtemp.Ok())
 		assert.Equal(t, "x", rtemp.Status())
 	})
+
+	t.Run("notImplemented", func(t *testing.T) {
+		rtemp := response{
+			status: "No handler found for this function",
+			err:    myCephError(-22),
+		}
+		if assert.False(t, rtemp.Ok()) {
+			err := rtemp.End()
+			assert.Error(t, err)
+			var n NotImplementedError
+			assert.True(t, errors.As(err, &n))
+			assert.Contains(t, err.Error(), "not implemented")
+		}
+	})
+}
+
+type myCephError int
+
+func (myCephError) Error() string {
+	return "oops"
+}
+
+func (e myCephError) ErrorCode() int {
+	return int(e)
 }
diff --git a/cephfs/admin/subvolume_octopus_test.go b/cephfs/admin/subvolume_octopus_test.go
deleted file mode 100644
index 836473e..0000000
--- a/cephfs/admin/subvolume_octopus_test.go
+++ /dev/null
@@ -1,93 +0,0 @@
-// +build octopus
-
-package admin
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-var sampleSubVolumeSnapshoInfo1 = []byte(`
-{
-    "created_at": "2020-09-11 17:40:12.035792",
-    "data_pool": "cephfs_data",
-    "has_pending_clones": "no",
-    "protected": "yes",
-    "size": 0
-}
-`)
-
-func TestParseSubVolumeSnapshotInfo(t *testing.T) {
-	R := newResponse
-	t.Run("error", func(t *testing.T) {
-		_, 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(R(nil, "unexpected!", nil))
-		assert.Error(t, err)
-	})
-	t.Run("badJSON", func(t *testing.T) {
-		_, err := parseSubVolumeSnapshotInfo(R([]byte("_XxXxX"), "", nil))
-		assert.Error(t, err)
-	})
-	t.Run("ok", func(t *testing.T) {
-		info, err := parseSubVolumeSnapshotInfo(R(sampleSubVolumeSnapshoInfo1, "", nil))
-		assert.NoError(t, err)
-		if assert.NotNil(t, info) {
-			assert.Equal(t, "cephfs_data", info.DataPool)
-			assert.EqualValues(t, 0, info.Size)
-			assert.Equal(t, 2020, info.CreatedAt.Year())
-			assert.Equal(t, "yes", info.Protected)
-			assert.Equal(t, "no", info.HasPendingClones)
-		}
-	})
-}
-
-func TestSubVolumeSnapshotInfo(t *testing.T) {
-	fsa := getFSAdmin(t)
-	volume := "cephfs"
-	group := "20000leagues"
-	subname := "poulp"
-	snapname1 := "t1"
-	snapname2 := "nope"
-
-	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, snapname1)
-	assert.NoError(t, err)
-	defer func() {
-		err := fsa.RemoveSubVolumeSnapshot(volume, group, subname, snapname1)
-		assert.NoError(t, err)
-	}()
-
-	sinfo, err := fsa.SubVolumeSnapshotInfo(volume, group, subname, snapname1)
-	assert.NoError(t, err)
-	assert.NotNil(t, sinfo)
-	assert.EqualValues(t, 0, sinfo.Size)
-	assert.Equal(t, "cephfs_data", sinfo.DataPool)
-	assert.GreaterOrEqual(t, 2020, sinfo.CreatedAt.Year())
-
-	sinfo, err = fsa.SubVolumeSnapshotInfo(volume, group, subname, snapname2)
-	assert.Error(t, err)
-	assert.Nil(t, sinfo)
-}
diff --git a/cephfs/admin/subvolume_test.go b/cephfs/admin/subvolume_test.go
index 84d489e..694d814 100644
--- a/cephfs/admin/subvolume_test.go
+++ b/cephfs/admin/subvolume_test.go
@@ -433,3 +433,86 @@ func TestSubVolumeSnapshots(t *testing.T) {
 		assert.Contains(t, snaps, snapname2)
 	})
 }
+
+var sampleSubVolumeSnapshoInfo1 = []byte(`
+{
+    "created_at": "2020-09-11 17:40:12.035792",
+    "data_pool": "cephfs_data",
+    "has_pending_clones": "no",
+    "protected": "yes",
+    "size": 0
+}
+`)
+
+func TestParseSubVolumeSnapshotInfo(t *testing.T) {
+	R := newResponse
+	t.Run("error", func(t *testing.T) {
+		_, 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(R(nil, "unexpected!", nil))
+		assert.Error(t, err)
+	})
+	t.Run("badJSON", func(t *testing.T) {
+		_, err := parseSubVolumeSnapshotInfo(R([]byte("_XxXxX"), "", nil))
+		assert.Error(t, err)
+	})
+	t.Run("ok", func(t *testing.T) {
+		info, err := parseSubVolumeSnapshotInfo(R(sampleSubVolumeSnapshoInfo1, "", nil))
+		assert.NoError(t, err)
+		if assert.NotNil(t, info) {
+			assert.Equal(t, "cephfs_data", info.DataPool)
+			assert.EqualValues(t, 0, info.Size)
+			assert.Equal(t, 2020, info.CreatedAt.Year())
+			assert.Equal(t, "yes", info.Protected)
+			assert.Equal(t, "no", info.HasPendingClones)
+		}
+	})
+}
+
+func TestSubVolumeSnapshotInfo(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "20000leagues"
+	subname := "poulp"
+	snapname1 := "t1"
+	snapname2 := "nope"
+
+	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, snapname1)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa.RemoveSubVolumeSnapshot(volume, group, subname, snapname1)
+		assert.NoError(t, err)
+	}()
+
+	sinfo, err := fsa.SubVolumeSnapshotInfo(volume, group, subname, snapname1)
+	assert.NoError(t, err)
+	assert.NotNil(t, sinfo)
+	assert.EqualValues(t, 0, sinfo.Size)
+	assert.Equal(t, "cephfs_data", sinfo.DataPool)
+	assert.GreaterOrEqual(t, 2020, sinfo.CreatedAt.Year())
+
+	sinfo, err = fsa.SubVolumeSnapshotInfo(volume, group, subname, snapname2)
+	assert.Error(t, err)
+	assert.Nil(t, sinfo)
+}
diff --git a/cephfs/admin/volume_octopus_test.go b/cephfs/admin/volume_octopus_test.go
deleted file mode 100644
index cd8fc85..0000000
--- a/cephfs/admin/volume_octopus_test.go
+++ /dev/null
@@ -1,52 +0,0 @@
-// +build octopus
-
-package admin
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestVolumeStatus(t *testing.T) {
-	fsa := getFSAdmin(t)
-
-	vs, err := fsa.VolumeStatus("cephfs")
-	assert.NoError(t, err)
-	assert.Contains(t, vs.MDSVersion, "version")
-}
-
-var sampleVolumeStatus1 = []byte(`
-{
-"clients": [{"clients": 1, "fs": "cephfs"}],
-"mds_version": "ceph version 15.2.4 (7447c15c6ff58d7fce91843b705a268a1917325c) octopus (stable)",
-"mdsmap": [{"dns": 76, "inos": 19, "name": "Z", "rank": 0, "rate": 0.0, "state": "active"}],
-"pools": [{"avail": 1017799872, "id": 2, "name": "cephfs_metadata", "type": "metadata", "used": 2204126}, {"avail": 1017799872, "id": 1, "name": "cephfs_data", "type": "data", "used": 0}]
-}
-`)
-
-func TestParseVolumeStatus(t *testing.T) {
-	R := newResponse
-	t.Run("error", func(t *testing.T) {
-		_, 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(R(nil, "unexpected!", nil))
-		assert.Error(t, err)
-	})
-	t.Run("badJSON", func(t *testing.T) {
-		_, err := parseVolumeStatus(R([]byte("_XxXxX"), "", nil))
-		assert.Error(t, err)
-	})
-	t.Run("ok", func(t *testing.T) {
-		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")
-			assert.Contains(t, s.MDSVersion, "octopus")
-		}
-	})
-}
diff --git a/cephfs/admin/volume_test.go b/cephfs/admin/volume_test.go
index e909e5a..0a1155c 100644
--- a/cephfs/admin/volume_test.go
+++ b/cephfs/admin/volume_test.go
@@ -176,3 +176,89 @@ func TestParseDumpToIdents(t *testing.T) {
 		assert.Nil(t, idents)
 	})
 }
+
+func TestVolumeStatus(t *testing.T) {
+	if !serverIsOctopus {
+		t.Skipf("can only execute on octopus servers")
+	}
+	fsa := getFSAdmin(t)
+
+	vs, err := fsa.VolumeStatus("cephfs")
+	assert.NoError(t, err)
+	assert.Contains(t, vs.MDSVersion, "version")
+}
+
+func TestVolumeStatusInvalid(t *testing.T) {
+	if !serverIsNautilus {
+		t.Skipf("can only excecute on nautilus servers")
+	}
+	fsa := getFSAdmin(t)
+
+	vs, err := fsa.VolumeStatus("cephfs")
+	assert.Error(t, err)
+	assert.Nil(t, vs)
+	var notImpl NotImplementedError
+	assert.True(t, errors.As(err, &notImpl))
+}
+
+var sampleVolumeStatus1 = []byte(`
+{
+"clients": [{"clients": 1, "fs": "cephfs"}],
+"mds_version": "ceph version 15.2.4 (7447c15c6ff58d7fce91843b705a268a1917325c) octopus (stable)",
+"mdsmap": [{"dns": 76, "inos": 19, "name": "Z", "rank": 0, "rate": 0.0, "state": "active"}],
+"pools": [{"avail": 1017799872, "id": 2, "name": "cephfs_metadata", "type": "metadata", "used": 2204126}, {"avail": 1017799872, "id": 1, "name": "cephfs_data", "type": "data", "used": 0}]
+}
+`)
+
+var sampleVolumeStatusTextJunk = []byte(`cephfs - 2 clients
+======
++------+--------+-----+---------------+-------+-------+
+| Rank | State  | MDS |    Activity   |  dns  |  inos |
++------+--------+-----+---------------+-------+-------+
+|  0   | active |  Z  | Reqs:   98 /s |  254  |  192  |
++------+--------+-----+---------------+-------+-------+
++-----------------+----------+-------+-------+
+|       Pool      |   type   |  used | avail |
++-----------------+----------+-------+-------+
+| cephfs_metadata | metadata | 62.1M |  910M |
+|   cephfs_data   |   data   |    0  |  910M |
++-----------------+----------+-------+-------+
++-------------+
+| Standby MDS |
++-------------+
++-------------+
+MDS version: ceph version 14.2.11 (f7fdb2f52131f54b891a2ec99d8205561242cdaf) nautilus (stable)
+`)
+
+func TestParseVolumeStatus(t *testing.T) {
+	R := newResponse
+	t.Run("error", func(t *testing.T) {
+		_, 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(R(nil, "unexpected!", nil))
+		assert.Error(t, err)
+	})
+	t.Run("badJSON", func(t *testing.T) {
+		_, err := parseVolumeStatus(R([]byte("_XxXxX"), "", nil))
+		assert.Error(t, err)
+	})
+	t.Run("ok", func(t *testing.T) {
+		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")
+			assert.Contains(t, s.MDSVersion, "octopus")
+		}
+	})
+	t.Run("notJSONfromServer", func(t *testing.T) {
+		_, err := parseVolumeStatus(R(sampleVolumeStatusTextJunk, "", nil))
+		if assert.Error(t, err) {
+			var notImpl NotImplementedError
+			assert.True(t, errors.As(err, &notImpl))
+		}
+	})
+
+}