diff --git a/cephfs/admin/bytecount.go b/cephfs/admin/bytecount.go
new file mode 100644
index 0000000..7f73956
--- /dev/null
+++ b/cephfs/admin/bytecount.go
@@ -0,0 +1,12 @@
+package admin
+
+// ByteCount represents the size of a volume in bytes.
+type ByteCount uint64
+
+// SI byte size constants. keep these private for now.
+const (
+	kibiByte ByteCount = 1024
+	mebiByte           = 1024 * kibiByte
+	gibiByte           = 1024 * mebiByte
+	tebiByte           = 1024 * gibiByte
+)
diff --git a/cephfs/admin/subvolume.go b/cephfs/admin/subvolume.go
new file mode 100644
index 0000000..4f83068
--- /dev/null
+++ b/cephfs/admin/subvolume.go
@@ -0,0 +1,95 @@
+package admin
+
+// this is the internal type used to create JSON for ceph.
+// See SubVolumeOptions for the type that users of the library
+// interact with.
+// note that the ceph json takes mode as a string.
+type subVolumeFields struct {
+	Prefix            string    `json:"prefix"`
+	Format            string    `json:"format"`
+	VolName           string    `json:"vol_name"`
+	GroupName         string    `json:"group_name,omitempty"`
+	SubName           string    `json:"sub_name"`
+	Size              ByteCount `json:"size,omitempty"`
+	Uid               int       `json:"uid,omitempty"`
+	Gid               int       `json:"gid,omitempty"`
+	Mode              string    `json:"mode,omitempty"`
+	PoolLayout        string    `json:"pool_layout,omitempty"`
+	NamespaceIsolated bool      `json:"namespace_isolated"`
+}
+
+// SubVolumeOptions are used to specify optional, non-identifying, values
+// to be used when creating a new subvolume.
+type SubVolumeOptions struct {
+	Size              ByteCount
+	Uid               int
+	Gid               int
+	Mode              int
+	PoolLayout        string
+	NamespaceIsolated bool
+}
+
+func (s *SubVolumeOptions) toFields(v, g, n string) *subVolumeFields {
+	return &subVolumeFields{
+		Prefix:            "fs subvolume create",
+		Format:            "json",
+		VolName:           v,
+		GroupName:         g,
+		SubName:           n,
+		Size:              s.Size,
+		Uid:               s.Uid,
+		Gid:               s.Gid,
+		Mode:              modeString(s.Mode, false),
+		PoolLayout:        s.PoolLayout,
+		NamespaceIsolated: s.NamespaceIsolated,
+	}
+}
+
+// NoGroup should be used when an optional subvolume group name is not
+// specified.
+const NoGroup = ""
+
+// CreateSubVolume sends a request to create a CephFS subvolume in a volume,
+// belonging to an optional subvolume group.
+func (fsa *FSAdmin) CreateSubVolume(volume, group, name string, o *SubVolumeOptions) error {
+	if o == nil {
+		o = &SubVolumeOptions{}
+	}
+	f := o.toFields(volume, group, name)
+	return checkEmptyResponseExpected(fsa.marshalMgrCommand(f))
+}
+
+// command:
+//   fs subvolume ls <vol_name> <group_name>
+
+// ListSubVolumes returns a list of subvolumes belonging to the volume and
+// optional subvolume group.
+func (fsa *FSAdmin) ListSubVolumes(volume, group string) ([]string, error) {
+	m := map[string]string{
+		"prefix":   "fs subvolume ls",
+		"vol_name": volume,
+		"format":   "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return parseListNames(fsa.marshalMgrCommand(m))
+}
+
+// command:
+//   fs subvolume rm <vol_name> <sub_name> <group_name> <force>
+
+// RemoveSubVolume will delete a CephFS subvolume in a volume and optional
+// subvolume group.
+func (fsa *FSAdmin) RemoveSubVolume(volume, group, name string) error {
+	m := map[string]string{
+		"prefix":   "fs subvolume rm",
+		"vol_name": volume,
+		"sub_name": name,
+		"format":   "json",
+	}
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+	return checkEmptyResponseExpected(fsa.marshalMgrCommand(m))
+}
diff --git a/cephfs/admin/subvolume_test.go b/cephfs/admin/subvolume_test.go
new file mode 100644
index 0000000..840dfee
--- /dev/null
+++ b/cephfs/admin/subvolume_test.go
@@ -0,0 +1,132 @@
+package admin
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func delay() {
+	// ceph seems to do this (partly?) async. So for now, we cheat
+	// and sleep a little to make subsequent tests more reliable
+	time.Sleep(50 * time.Millisecond)
+}
+
+func TestCreateSubVolume(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	type gn struct {
+		group string
+		name  string
+	}
+	created := []gn{}
+	defer func() {
+		for _, c := range created {
+			err := fsa.RemoveSubVolume(volume, c.group, c.name)
+			assert.NoError(t, err)
+			delay()
+			if c.group != NoGroup {
+				err := fsa.RemoveSubVolumeGroup(volume, c.group)
+				assert.NoError(t, err)
+			}
+		}
+	}()
+
+	t.Run("simple", func(t *testing.T) {
+		subname := "SubVol1"
+		err := fsa.CreateSubVolume(volume, NoGroup, subname, nil)
+		assert.NoError(t, err)
+		created = append(created, gn{NoGroup, subname})
+
+		lsv, err := fsa.ListSubVolumes(volume, NoGroup)
+		assert.NoError(t, err)
+		assert.GreaterOrEqual(t, len(lsv), 1)
+		assert.Contains(t, lsv, subname)
+	})
+
+	t.Run("options", func(t *testing.T) {
+		subname := "SubVol2"
+		o := &SubVolumeOptions{
+			Mode: 0777,
+			Uid:  200,
+			Gid:  200,
+		}
+		err := fsa.CreateSubVolume(volume, NoGroup, subname, o)
+		assert.NoError(t, err)
+		created = append(created, gn{NoGroup, subname})
+
+		lsv, err := fsa.ListSubVolumes(volume, NoGroup)
+		assert.NoError(t, err)
+		assert.GreaterOrEqual(t, len(lsv), 1)
+		assert.Contains(t, lsv, subname)
+	})
+
+	t.Run("withGroup", func(t *testing.T) {
+		group := "withGroup1"
+		subname := "SubVol3"
+
+		err := fsa.CreateSubVolumeGroup(volume, group, nil)
+		assert.NoError(t, err)
+
+		err = fsa.CreateSubVolume(volume, group, subname, nil)
+		assert.NoError(t, err)
+		created = append(created, gn{group, subname})
+
+		lsv, err := fsa.ListSubVolumes(volume, group)
+		assert.NoError(t, err)
+		assert.GreaterOrEqual(t, len(lsv), 1)
+		assert.Contains(t, lsv, subname)
+	})
+
+	t.Run("groupAndOptions", func(t *testing.T) {
+		group := "withGroup2"
+		subname := "SubVol4"
+		err := fsa.CreateSubVolumeGroup(volume, group, nil)
+		assert.NoError(t, err)
+
+		o := &SubVolumeOptions{
+			Size: 5 * gibiByte,
+			Mode: 0777,
+			Uid:  200,
+			Gid:  200,
+		}
+		err = fsa.CreateSubVolume(volume, group, subname, o)
+		assert.NoError(t, err)
+		created = append(created, gn{group, subname})
+
+		lsv, err := fsa.ListSubVolumes(volume, group)
+		assert.NoError(t, err)
+		assert.GreaterOrEqual(t, len(lsv), 1)
+		assert.Contains(t, lsv, subname)
+	})
+}
+
+func TestRemoveSubVolume(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+
+	lsv, err := fsa.ListSubVolumes(volume, NoGroup)
+	assert.NoError(t, err)
+	beforeCount := len(lsv)
+
+	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)
+
+	err = fsa.RemoveSubVolume(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)
+	}
+}