cephfs admin: return not-impl errors if ceph returns text for volume status

Return not implemented errors when "older" versions of ceph return
formatted text for output instead of JSON, even tho we asked for json.

Signed-off-by: John Mulligan <jmulligan@redhat.com>

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) {
@@ -88,8 +94,19 @@ type VolumeStatus struct {
 
 func parseVolumeStatus(res response) (*VolumeStatus, error) {
 	var vs VolumeStatus
-	err := res.noStatus().unmarshal(&vs).End()
-	return &vs, err
+	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.

cephfs/admin/volume_octopus_test.go

@@ -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")
-		}
-	})
-}

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))
+		}
+	})
+
+}