cephfs: add subvolume metadata APIs

cephfs/admin/fsadmin.go

@@ -93,6 +93,15 @@ func parseListNames(res response) ([]string, error) {
 	return vl, nil
 }
 
+func parseListKeyValues(res response) (map[string]string, error) {
+	var x map[string]string
+	if err := res.NoStatus().Unmarshal(&x).End(); err != nil {
+		return nil, err
+	}
+
+	return x, nil
+}
+
 // parsePathResponse returns a cleaned up path from requests that get a path
 // unless an error is encountered, then an error is returned.
 func parsePathResponse(res response) (string, error) {

cephfs/admin/metadata.go

@@ -0,0 +1,104 @@
+//go:build !(nautilus || octopus) && ceph_preview && ceph_ci_untested
+// +build !nautilus,!octopus,ceph_preview,ceph_ci_untested
+
+package admin
+
+import "C"
+
+// GetMetadata gets custom metadata on the subvolume in a volume belonging to
+// an optional subvolume group based on provided key name.
+//
+// Similar To:
+//  ceph fs subvolume metadata get <vol_name> <sub_name> <key_name> [--group_name <subvol_group_name>]
+func (fsa *FSAdmin) GetMetadata(volume, group, subvolume, key string) (string, error) {
+	m := map[string]string{
+		"prefix":   "fs subvolume metadata get",
+		"format":   "json",
+		"vol_name": volume,
+		"sub_name": subvolume,
+		"key_name": key,
+	}
+
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+
+	return parsePathResponse(fsa.marshalMgrCommand(m))
+}
+
+// SetMetadata sets custom metadata on the subvolume in a volume belonging to
+// an optional subvolume group as a key-value pair.
+//
+// Similar To:
+//  ceph fs subvolume metadata set <vol_name> <sub_name> <key_name> <value> [--group_name <subvol_group_name>]
+func (fsa *FSAdmin) SetMetadata(volume, group, subvolume, key, value string) error {
+	m := map[string]string{
+		"prefix":   "fs subvolume metadata set",
+		"format":   "json",
+		"vol_name": volume,
+		"sub_name": subvolume,
+		"key_name": key,
+		"value":    value,
+	}
+
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+
+	return fsa.marshalMgrCommand(m).NoData().End()
+}
+
+// RemoveMetadata removes custom metadata set on the subvolume in a volume
+// belonging to an optional subvolume group using the metadata key.
+//
+// Similar To:
+//  ceph fs subvolume metadata rm <vol_name> <sub_name> <key_name> [--group_name <subvol_group_name>]
+func (fsa *FSAdmin) RemoveMetadata(volume, group, subvolume, key string) error {
+	return fsa.rmSubVolumeMetadata(volume, group, subvolume, key, commonRmFlags{})
+}
+
+// ForceRemoveMetadata attempt to forcefully remove custom metadata set on
+// the subvolume in a volume belonging to an optional subvolume group using
+// the metadata key.
+//
+// Similar To:
+//  ceph fs subvolume metadata rm <vol_name> <sub_name> <key_name> [--group_name <subvol_group_name>] --force
+func (fsa *FSAdmin) ForceRemoveMetadata(volume, group, subvolume, key string) error {
+	return fsa.rmSubVolumeMetadata(volume, group, subvolume, key, commonRmFlags{force: true})
+}
+
+func (fsa *FSAdmin) rmSubVolumeMetadata(volume, group, subvolume, key string, o commonRmFlags) error {
+	m := map[string]string{
+		"prefix":   "fs subvolume metadata rm",
+		"format":   "json",
+		"vol_name": volume,
+		"sub_name": subvolume,
+		"key_name": key,
+	}
+
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+
+	return fsa.marshalMgrCommand(mergeFlags(m, o)).NoData().End()
+}
+
+// ListMetadata lists custom metadata (key-value pairs) set on the subvolume
+// in a volume belonging to an optional subvolume group.
+//
+// Similar To:
+//  ceph fs subvolume metadata ls <vol_name> <sub_name> [--group_name <subvol_group_name>]
+func (fsa *FSAdmin) ListMetadata(volume, group, subvolume string) (map[string]string, error) {
+	m := map[string]string{
+		"prefix":   "fs subvolume metadata ls",
+		"format":   "json",
+		"vol_name": volume,
+		"sub_name": subvolume,
+	}
+
+	if group != NoGroup {
+		m["group_name"] = group
+	}
+
+	return parseListKeyValues(fsa.marshalMgrCommand(m))
+}

cephfs/admin/metadata_test.go

@@ -0,0 +1,155 @@
+//go:build !(nautilus || octopus) && ceph_preview && ceph_ci_untested
+// +build !nautilus,!octopus,ceph_preview,ceph_ci_untested
+
+package admin
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSetMetadata(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "group"
+	subname := "subVol"
+	key := "hi"
+	value := "hello"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	err = fsa.CreateSubVolume(volume, group, subname, nil)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key, value)
+	assert.NoError(t, err)
+
+	err = fsa.RemoveSubVolume(volume, group, subname)
+	assert.NoError(t, err)
+	err = fsa.RemoveSubVolumeGroup(volume, group)
+	assert.NoError(t, err)
+}
+
+func TestGetMetadata(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "group"
+	subname := "subVol"
+	key := "hi"
+	value := "hello"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	err = fsa.CreateSubVolume(volume, group, subname, nil)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key, value)
+	assert.NoError(t, err)
+
+	metaValue, err := fsa.GetMetadata(volume, group, subname, key)
+	assert.NoError(t, err)
+	assert.Equal(t, metaValue, value)
+
+	err = fsa.RemoveSubVolume(volume, group, subname)
+	assert.NoError(t, err)
+	err = fsa.RemoveSubVolumeGroup(volume, group)
+	assert.NoError(t, err)
+}
+
+func TestRemoveMetadata(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "group"
+	subname := "subVol"
+	key := "hi"
+	value := "hello"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	err = fsa.CreateSubVolume(volume, group, subname, nil)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key, value)
+	assert.NoError(t, err)
+
+	metaValue, err := fsa.GetMetadata(volume, group, subname, key)
+	assert.NoError(t, err)
+	assert.Equal(t, metaValue, value)
+
+	err = fsa.RemoveMetadata(volume, group, subname, key)
+	assert.NoError(t, err)
+
+	metaValue, err = fsa.GetMetadata(volume, group, subname, key)
+	assert.Error(t, err)
+
+	err = fsa.RemoveSubVolume(volume, group, subname)
+	assert.NoError(t, err)
+	err = fsa.RemoveSubVolumeGroup(volume, group)
+	assert.NoError(t, err)
+}
+
+func TestForceRemoveMetadata(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "group"
+	subname := "subVol"
+	key := "hi"
+	value := "hello"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	err = fsa.CreateSubVolume(volume, group, subname, nil)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key, value)
+	assert.NoError(t, err)
+
+	metaValue, err := fsa.GetMetadata(volume, group, subname, key)
+	assert.NoError(t, err)
+	assert.Equal(t, metaValue, value)
+
+	err = fsa.ForceRemoveMetadata(volume, group, subname, key)
+	assert.NoError(t, err)
+
+	metaValue, err = fsa.GetMetadata(volume, group, subname, key)
+	assert.Error(t, err)
+
+	err = fsa.RemoveSubVolume(volume, group, subname)
+	assert.NoError(t, err)
+	err = fsa.RemoveSubVolumeGroup(volume, group)
+	assert.NoError(t, err)
+}
+
+func TestListMetadata(t *testing.T) {
+	fsa := getFSAdmin(t)
+	volume := "cephfs"
+	group := "group"
+	subname := "subVol"
+	key1 := "hi1"
+	value1 := "hello1"
+	key2 := "hi2"
+	value2 := "hello2"
+
+	err := fsa.CreateSubVolumeGroup(volume, group, nil)
+	assert.NoError(t, err)
+	err = fsa.CreateSubVolume(volume, group, subname, nil)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key1, value1)
+	assert.NoError(t, err)
+
+	err = fsa.SetMetadata(volume, group, subname, key2, value2)
+	assert.NoError(t, err)
+
+	metaList, err := fsa.ListMetadata(volume, group, subname)
+	assert.NoError(t, err)
+	assert.GreaterOrEqual(t, len(metaList), 2)
+	assert.Contains(t, metaList, key1)
+	assert.Contains(t, metaList, key2)
+
+	err = fsa.RemoveSubVolume(volume, group, subname)
+	assert.NoError(t, err)
+	err = fsa.RemoveSubVolumeGroup(volume, group)
+	assert.NoError(t, err)
+}