core: parse subvolumegroup bytes_quota reported as "infinite"

Ceph reports bytes_quota in `ceph fs subvolumegroup info` as the JSON
string "infinite" when no quota is set, and as a number otherwise. The
subvolumeGroupInfo.BytesQuota field is typed int64, so json.Unmarshal
fails with "cannot unmarshal string into Go value of type int64" for any
subvolume group without a quota (the default).

That unmarshal error is returned from getCephFSSubVolumeGroupInfo and
swallowed in CreateCephFSSubVolumeGroup: exec.ExitStatus reports ok=false
for the wrapped error, so the non-ENOENT guard does not fire and the
dedicated resize path (which enforces --no-shrink) is skipped.

Add an UnmarshalJSON on subvolumeGroupInfo that accepts either a number
or the "infinite" sentinel, mapping "infinite" to 0 (unlimited) so the
existing CmpInt64 comparison keeps working. Add a table-driven test
covering the finite and infinite cases.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

pkg/daemon/ceph/client/subvolumegroup.go

@@ -96,6 +96,38 @@ type subvolumeGroupInfo struct {
 	DataPool   string `json:"data_pool"`
 }
 
+// UnmarshalJSON handles the bytes_quota field, which Ceph reports as the JSON
+// string "infinite" when no quota is set (see Ceph's Group.info) and as a
+// number otherwise. The "infinite" sentinel is mapped to 0 (unlimited).
+func (s *subvolumeGroupInfo) UnmarshalJSON(data []byte) error {
+	type subvolumeGroupInfoAlias subvolumeGroupInfo
+	aux := &struct {
+		BytesQuota json.RawMessage `json:"bytes_quota"`
+		*subvolumeGroupInfoAlias
+	}{
+		subvolumeGroupInfoAlias: (*subvolumeGroupInfoAlias)(s),
+	}
+	if err := json.Unmarshal(data, aux); err != nil {
+		return err
+	}
+
+	if len(aux.BytesQuota) == 0 || string(aux.BytesQuota) == "null" {
+		s.BytesQuota = 0
+		return nil
+	}
+
+	var quotaStr string
+	if err := json.Unmarshal(aux.BytesQuota, &quotaStr); err == nil {
+		if quotaStr == "infinite" {
+			s.BytesQuota = 0
+			return nil
+		}
+		return errors.Errorf("unexpected bytes_quota value %q", quotaStr)
+	}
+
+	return json.Unmarshal(aux.BytesQuota, &s.BytesQuota)
+}
+
 // getCephFSSubVolumeGroupInfo get subvolumegroup info of the group name.
 // volName is the name of the Ceph FS volume, the same as the CephFilesystem CR name.
 func getCephFSSubVolumeGroupInfo(context *clusterd.Context, clusterInfo *ClusterInfo, volName, groupName string) (*subvolumeGroupInfo, error) {

pkg/daemon/ceph/client/subvolumegroup_test.go

@@ -19,7 +19,10 @@ package client
 import (
 	"testing"
 
+	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -79,3 +82,44 @@ func TestValidatePinningValues(t *testing.T) {
 	err = validatePinningValues(testData1)
 	assert.NoError(t, err)
 }
+
+func TestGetCephFSSubVolumeGroupInfo(t *testing.T) {
+	tests := []struct {
+		name             string
+		output           string
+		expectedQuota    int64
+		expectedDataPool string
+	}{
+		{
+			name:             "finite quota",
+			output:           `{"bytes_quota":1099511627776,"bytes_used":0,"data_pool":"myfs-replicated"}`,
+			expectedQuota:    1099511627776,
+			expectedDataPool: "myfs-replicated",
+		},
+		{
+			// Ceph reports bytes_quota as the string "infinite" when no quota is set.
+			name:             "infinite quota",
+			output:           `{"bytes_quota":"infinite","bytes_used":0,"data_pool":"myfs-replicated"}`,
+			expectedQuota:    0,
+			expectedDataPool: "myfs-replicated",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			executor := &exectest.MockExecutor{}
+			executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "fs" && args[1] == "subvolumegroup" && args[2] == "info" {
+					return tt.output, nil
+				}
+				return "", errors.Errorf("unexpected ceph command %q", args)
+			}
+
+			svgInfo, err := getCephFSSubVolumeGroupInfo(&clusterd.Context{Executor: executor}, AdminTestClusterInfo("mycluster"), "myfs", "csi")
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expectedQuota, svgInfo.BytesQuota)
+			assert.Equal(t, tt.expectedDataPool, svgInfo.DataPool)
+		})
+	}
+}