ceph: fix improper json parsing

pkg/operator/ceph/object/admin.go

@@ -18,6 +18,7 @@ package object
 
 import (
 	"fmt"
+	"regexp"
 	"time"
 
 	"github.com/pkg/errors"
@@ -56,9 +57,20 @@ func NewMultisiteContext(context *clusterd.Context, clusterInfo *client.ClusterI
 	return objContext, nil
 }
 
+func extractJSON(output string) (string, error) {
+	// `radosgw-admin` sometimes leaves logs to stderr even if it succeeds.
+	// So we should skip them if parsing output as json.
+	pattern := regexp.MustCompile("(?ms)^{.*")
+	match := pattern.Find([]byte(output))
+	if match == nil {
+		return "", errors.Errorf("didn't contain json. %s", output)
+	}
+	return string(match), nil
+}
+
 // RunAdminCommandNoMultisite is for running radosgw-admin commands in scenarios where an object-store has not been created yet or for commands on the realm or zonegroup (ex: radosgw-admin zonegroup get)
 // This function times out after a fixed interval if no response is received.
-func RunAdminCommandNoMultisite(c *Context, args ...string) (string, error) {
+func RunAdminCommandNoMultisite(c *Context, expectJSON bool, args ...string) (string, error) {
 	command, args := client.FinalizeCephCommandArgs("radosgw-admin", c.clusterInfo, args, c.Context.ConfigDir)
 
 	timeout, err := time.ParseDuration(fmt.Sprintf("%ss", client.CephConnectionTimeout))
@@ -71,12 +83,19 @@ func RunAdminCommandNoMultisite(c *Context, args ...string) (string, error) {
 	if err != nil {
 		return output, err
 	}
+	if expectJSON {
+		match, err := extractJSON(output)
+		if err != nil {
+			return output, errors.Wrap(err, "failed to parse as JSON")
+		}
+		output = match
+	}
 
 	return output, nil
 }
 
 // This function is for running radosgw-admin commands in scenarios where an object-store has been created and the Context has been updated with the appropriate realm, zone group, and zone.
-func runAdminCommand(c *Context, args ...string) (string, error) {
+func runAdminCommand(c *Context, expectJSON bool, args ...string) (string, error) {
 	// If the objectStoreName is not passed in the storage class
 	// This means we are pointing to an external cluster so these commands are not needed
 	// simply because the external cluster mode does not support that yet
@@ -91,8 +110,8 @@ func runAdminCommand(c *Context, args ...string) (string, error) {
 			fmt.Sprintf("--rgw-zone=%s", c.Zone),
 		}
 
-		return RunAdminCommandNoMultisite(c, append(args, options...)...)
+		return RunAdminCommandNoMultisite(c, expectJSON, append(args, options...)...)
 	}
 
-	return RunAdminCommandNoMultisite(c, args...)
+	return RunAdminCommandNoMultisite(c, expectJSON, args...)
 }

pkg/operator/ceph/object/admin_test.go

@@ -0,0 +1,48 @@
+/*
+Copyright 2021 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package object
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExtractJson(t *testing.T) {
+	s := "invalid json"
+	_, err := extractJSON(s)
+	assert.Error(t, err)
+
+	s = `{"test": "test"}`
+	match, err := extractJSON(s)
+	assert.NoError(t, err)
+	assert.True(t, json.Valid([]byte(match)))
+
+	s = `this line can't be parsed as json
+{"test": "test"}`
+	match, err = extractJSON(s)
+	assert.NoError(t, err)
+	assert.True(t, json.Valid([]byte(match)))
+
+	s = `this line can't be parsed as json
+{"test":
+"test"}`
+	match, err = extractJSON(s)
+	assert.NoError(t, err)
+	assert.True(t, json.Valid([]byte(match)))
+}

pkg/operator/ceph/object/bucket.go

@@ -80,6 +80,7 @@ func bucketStatsFromRGW(stats rgwBucketStats) ObjectBucketStats {
 
 func GetBucketStats(c *Context, bucketName string) (*ObjectBucketStats, bool, error) {
 	result, err := runAdminCommand(c,
+		true,
 		"bucket",
 		"stats",
 		"--bucket", bucketName)
@@ -104,6 +105,7 @@ func GetBucketStats(c *Context, bucketName string) (*ObjectBucketStats, bool, er
 
 func GetBucketsStats(c *Context) (map[string]ObjectBucketStats, error) {
 	result, err := runAdminCommand(c,
+		true,
 		"bucket",
 		"stats")
 	if err != nil {
@@ -126,6 +128,7 @@ func GetBucketsStats(c *Context) (map[string]ObjectBucketStats, error) {
 
 func getBucketMetadata(c *Context, bucket string) (*ObjectBucketMetadata, bool, error) {
 	result, err := runAdminCommand(c,
+		false,
 		"metadata",
 		"get",
 		"bucket:"+bucket)
@@ -136,15 +139,19 @@ func getBucketMetadata(c *Context, bucket string) (*ObjectBucketMetadata, bool,
 	if strings.Contains(result, "can't get key") {
 		return nil, true, errors.New("not found")
 	}
+	match, err := extractJSON(result)
+	if err != nil {
+		return nil, false, errors.Wrapf(err, "failed to read buckets list result=%s", result)
+	}
 
 	var s struct {
 		Data struct {
 			Owner        string `json:"owner"`
 			CreationTime string `json:"creation_time"`
 		} `json:"data"`
 	}
-	if err := json.Unmarshal([]byte(result), &s); err != nil {
-		return nil, false, errors.Wrapf(err, "failed to read buckets list result=%s", result)
+	if err := json.Unmarshal([]byte(match), &s); err != nil {
+		return nil, false, errors.Wrapf(err, "failed to read buckets list result=%s", match)
 	}
 
 	timeParser := octopusAndAfterTime
@@ -216,7 +223,7 @@ func DeleteObjectBucket(c *Context, bucketName string, purge bool) (int, error)
 		options = append(options, "--purge-objects")
 	}
 
-	result, err := runAdminCommand(c, options...)
+	result, err := runAdminCommand(c, false, options...)
 	if err != nil {
 		return RGWErrorUnknown, errors.Wrap(err, "failed to delete bucket")
 	}

pkg/operator/ceph/object/controller.go

@@ -412,7 +412,7 @@ func (r *ReconcileCephObjectStore) reconcileCephZone(store *cephv1.CephObjectSto
 	zoneArg := fmt.Sprintf("--rgw-zone=%s", store.Spec.Zone.Name)
 	objContext := NewContext(r.context, r.clusterInfo, store.Name)
 
-	_, err := RunAdminCommandNoMultisite(objContext, "zone", "get", realmArg, zoneGroupArg, zoneArg)
+	_, err := RunAdminCommandNoMultisite(objContext, true, "zone", "get", realmArg, zoneGroupArg, zoneArg)
 	if err != nil {
 		if code, ok := exec.ExitStatus(err); ok && code == int(syscall.ENOENT) {
 			return waitForRequeueIfObjectStoreNotReady, errors.Wrapf(err, "ceph zone %q not found", store.Spec.Zone.Name)

pkg/operator/ceph/object/objectstore.go

@@ -119,7 +119,7 @@ func removeObjectStoreFromMultisite(objContext *Context, spec cephv1.ObjectStore
 
 	zoneGroupIsMaster := false
 	if zoneIsMaster {
-		_, err = RunAdminCommandNoMultisite(objContext, "zonegroup", "modify", realmArg, zoneGroupArg, endpointArg)
+		_, err = RunAdminCommandNoMultisite(objContext, false, "zonegroup", "modify", realmArg, zoneGroupArg, endpointArg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to remove object store %q endpoint from rgw zone group %q", objContext.Name, objContext.ZoneGroup)
 		}
@@ -132,7 +132,7 @@ func removeObjectStoreFromMultisite(objContext *Context, spec cephv1.ObjectStore
 		}
 	}
 
-	_, err = runAdminCommand(objContext, "zone", "modify", endpointArg)
+	_, err = runAdminCommand(objContext, false, "zone", "modify", endpointArg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to remove object store %q endpoint from rgw zone %q", objContext.Name, spec.Zone.Name)
 	}
@@ -143,7 +143,7 @@ func removeObjectStoreFromMultisite(objContext *Context, spec cephv1.ObjectStore
 	}
 
 	// the period will help notify other zones of changes if there are multi-zones
-	_, err = runAdminCommand(objContext, "period", "update", "--commit")
+	_, err = runAdminCommand(objContext, false, "period", "update", "--commit")
 	if err != nil {
 		return errors.Wrap(err, "failed to update period after removing an endpoint from the zone")
 	}
@@ -216,7 +216,7 @@ func checkZoneIsMaster(objContext *Context) (bool, error) {
 	zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", objContext.ZoneGroup)
 	zoneArg := fmt.Sprintf("--rgw-zone=%s", objContext.Zone)
 
-	zoneGroupJson, err := RunAdminCommandNoMultisite(objContext, "zonegroup", "get", realmArg, zoneGroupArg)
+	zoneGroupJson, err := RunAdminCommandNoMultisite(objContext, true, "zonegroup", "get", realmArg, zoneGroupArg)
 	if err != nil {
 		return false, errors.Wrap(err, "failed to get rgw zone group")
 	}
@@ -226,7 +226,7 @@ func checkZoneIsMaster(objContext *Context) (bool, error) {
 	}
 	logger.Debugf("got master zone ID for zone group %v", objContext.ZoneGroup)
 
-	zoneOutput, err := RunAdminCommandNoMultisite(objContext, "zone", "get", realmArg, zoneGroupArg, zoneArg)
+	zoneOutput, err := RunAdminCommandNoMultisite(objContext, true, "zone", "get", realmArg, zoneGroupArg, zoneArg)
 	if err != nil {
 		return false, errors.Wrap(err, "failed to get rgw zone")
 	}
@@ -250,7 +250,7 @@ func checkZoneGroupIsMaster(objContext *Context) (bool, error) {
 	realmArg := fmt.Sprintf("--rgw-realm=%s", objContext.Realm)
 	zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", objContext.ZoneGroup)
 
-	zoneGroupOutput, err := RunAdminCommandNoMultisite(objContext, "zonegroup", "get", realmArg, zoneGroupArg)
+	zoneGroupOutput, err := RunAdminCommandNoMultisite(objContext, true, "zonegroup", "get", realmArg, zoneGroupArg)
 	if err != nil {
 		return false, errors.Wrap(err, "failed to get rgw zone group")
 	}
@@ -310,7 +310,7 @@ func getZoneEndpoints(objContext *Context, serviceEndpoint string) ([]string, er
 	realmArg := fmt.Sprintf("--rgw-realm=%s", objContext.Realm)
 	zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", objContext.ZoneGroup)
 
-	zoneGroupOutput, err := RunAdminCommandNoMultisite(objContext, "zonegroup", "get", realmArg, zoneGroupArg)
+	zoneGroupOutput, err := RunAdminCommandNoMultisite(objContext, true, "zonegroup", "get", realmArg, zoneGroupArg)
 	if err != nil {
 		return []string{}, errors.Wrap(err, "failed to get rgw zone group")
 	}
@@ -343,11 +343,11 @@ func createMultisite(objContext *Context, endpointArg string) error {
 
 	updatePeriod := false
 	// create the realm if it doesn't exist yet
-	output, err := RunAdminCommandNoMultisite(objContext, "realm", "get", realmArg)
+	output, err := RunAdminCommandNoMultisite(objContext, true, "realm", "get", realmArg)
 	if err != nil {
 		if code, ok := exec.ExitStatus(err); ok && code == int(syscall.ENOENT) {
 			updatePeriod = true
-			output, err = RunAdminCommandNoMultisite(objContext, "realm", "create", realmArg)
+			output, err = RunAdminCommandNoMultisite(objContext, false, "realm", "create", realmArg)
 			if err != nil {
 				return errors.Wrapf(err, "failed to create ceph realm %q, for reason %q", objContext.ZoneGroup, output)
 			}
@@ -358,11 +358,11 @@ func createMultisite(objContext *Context, endpointArg string) error {
 	}
 
 	// create the zonegroup if it doesn't exist yet
-	output, err = RunAdminCommandNoMultisite(objContext, "zonegroup", "get", realmArg, zoneGroupArg)
+	output, err = RunAdminCommandNoMultisite(objContext, true, "zonegroup", "get", realmArg, zoneGroupArg)
 	if err != nil {
 		if code, ok := exec.ExitStatus(err); ok && code == int(syscall.ENOENT) {
 			updatePeriod = true
-			output, err = RunAdminCommandNoMultisite(objContext, "zonegroup", "create", "--master", realmArg, zoneGroupArg, endpointArg)
+			output, err = RunAdminCommandNoMultisite(objContext, false, "zonegroup", "create", "--master", realmArg, zoneGroupArg, endpointArg)
 			if err != nil {
 				return errors.Wrapf(err, "failed to create ceph zone group %q, for reason %q", objContext.ZoneGroup, output)
 			}
@@ -373,11 +373,11 @@ func createMultisite(objContext *Context, endpointArg string) error {
 	}
 
 	// create the zone if it doesn't exist yet
-	output, err = runAdminCommand(objContext, "zone", "get")
+	output, err = runAdminCommand(objContext, true, "zone", "get")
 	if err != nil {
 		if code, ok := exec.ExitStatus(err); ok && code == int(syscall.ENOENT) {
 			updatePeriod = true
-			output, err = runAdminCommand(objContext, "zone", "create", "--master", endpointArg)
+			output, err = runAdminCommand(objContext, false, "zone", "create", "--master", endpointArg)
 			if err != nil {
 				return errors.Wrapf(err, "failed to create ceph zone %q, for reason %q", objContext.Zone, output)
 			}
@@ -389,7 +389,7 @@ func createMultisite(objContext *Context, endpointArg string) error {
 
 	if updatePeriod {
 		// the period will help notify other zones of changes if there are multi-zones
-		_, err := runAdminCommand(objContext, "period", "update", "--commit")
+		_, err := runAdminCommand(objContext, false, "period", "update", "--commit")
 		if err != nil {
 			return errors.Wrap(err, "failed to update period")
 		}
@@ -415,7 +415,7 @@ func joinMultisite(objContext *Context, endpointArg, zoneEndpoints, namespace st
 
 	if zoneIsMaster {
 		// endpoints that are part of a master zone are supposed to be the endpoints for a zone group
-		_, err := RunAdminCommandNoMultisite(objContext, "zonegroup", "modify", realmArg, zoneGroupArg, endpointArg)
+		_, err := RunAdminCommandNoMultisite(objContext, false, "zonegroup", "modify", realmArg, zoneGroupArg, endpointArg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to add object store %q in rgw zone group %q", objContext.Name, objContext.ZoneGroup)
 		}
@@ -427,14 +427,14 @@ func joinMultisite(objContext *Context, endpointArg, zoneEndpoints, namespace st
 			return errors.Wrapf(err, "failed to find out whether zone group %q in is the master zone group", objContext.ZoneGroup)
 		}
 	}
-	_, err = RunAdminCommandNoMultisite(objContext, "zone", "modify", realmArg, zoneGroupArg, zoneArg, endpointArg)
+	_, err = RunAdminCommandNoMultisite(objContext, false, "zone", "modify", realmArg, zoneGroupArg, zoneArg, endpointArg)
 	if err != nil {
 		return errors.Wrapf(err, "failed to add object store %q in rgw zone %q", objContext.Name, objContext.Zone)
 	}
 	logger.Debugf("endpoints for zone %q are now %q", objContext.Zone, zoneEndpoints)
 
 	// the period will help notify other zones of changes if there are multi-zones
-	_, err = RunAdminCommandNoMultisite(objContext, "period", "update", "--commit", realmArg, zoneGroupArg, zoneArg)
+	_, err = RunAdminCommandNoMultisite(objContext, false, "period", "update", "--commit", realmArg, zoneGroupArg, zoneArg)
 	if err != nil {
 		return errors.Wrap(err, "failed to update period")
 	}
@@ -458,7 +458,7 @@ func createSystemUser(objContext *Context, namespace string) error {
 	zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", objContext.ZoneGroup)
 	zoneArg := fmt.Sprintf("--rgw-zone=%s", objContext.Zone)
 
-	output, err := RunAdminCommandNoMultisite(objContext, "user", "info", uidArg)
+	output, err := RunAdminCommandNoMultisite(objContext, false, "user", "info", uidArg)
 	if err == nil {
 		logger.Debugf("realm system user %q has already been created", uid)
 		return nil
@@ -473,7 +473,7 @@ func createSystemUser(objContext *Context, namespace string) error {
 		logger.Debugf("found keys to create realm system user %v", uid)
 		systemArg := "--system"
 		displayNameArg := fmt.Sprintf("--display-name=%s.user", objContext.Realm)
-		output, err = RunAdminCommandNoMultisite(objContext, "user", "create", realmArg, zoneGroupArg, zoneArg, uidArg, displayNameArg, accessKeyArg, secretKeyArg, systemArg)
+		output, err = RunAdminCommandNoMultisite(objContext, false, "user", "create", realmArg, zoneGroupArg, zoneArg, uidArg, displayNameArg, accessKeyArg, secretKeyArg, systemArg)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create realm system user %q for reason: %q", uid, output)
 		}
@@ -523,17 +523,17 @@ func deleteRealm(context *Context) error {
 	//  <name>
 	realmArg := fmt.Sprintf("--rgw-realm=%s", context.Name)
 	zoneGroupArg := fmt.Sprintf("--rgw-zonegroup=%s", context.Name)
-	_, err := RunAdminCommandNoMultisite(context, "realm", "delete", realmArg)
+	_, err := RunAdminCommandNoMultisite(context, false, "realm", "delete", realmArg)
 	if err != nil {
 		logger.Warningf("failed to delete rgw realm %q. %v", context.Name, err)
 	}
 
-	_, err = RunAdminCommandNoMultisite(context, "zonegroup", "delete", realmArg, zoneGroupArg)
+	_, err = RunAdminCommandNoMultisite(context, false, "zonegroup", "delete", realmArg, zoneGroupArg)
 	if err != nil {
 		logger.Warningf("failed to delete rgw zonegroup %q. %v", context.Name, err)
 	}
 
-	_, err = runAdminCommand(context, "zone", "delete")
+	_, err = runAdminCommand(context, false, "zone", "delete")
 	if err != nil {
 		logger.Warningf("failed to delete rgw zone %q. %v", context.Name, err)
 	}
@@ -562,7 +562,7 @@ func DecodeZoneGroupConfig(data string) (zoneGroupType, error) {
 }
 
 func getObjectStores(context *Context) ([]string, error) {
-	output, err := RunAdminCommandNoMultisite(context, "realm", "list")
+	output, err := RunAdminCommandNoMultisite(context, true, "realm", "list")
 	if err != nil {
 		// exit status 2 indicates the object store does not exist, so return nothing
 		if strings.Index(err.Error(), "exit status 2") == 0 {

pkg/operator/ceph/object/objectstore_test.go

@@ -67,9 +67,15 @@ func TestReconcileRealm(t *testing.T) {
 		logger.Infof("Execute: %s %v", command, args)
 		return idResponse, nil
 	}
+	executorFuncTimeout := func(timeout time.Duration, command string, args ...string) (string, error) {
+		testResponse := `{"id": "test-id"}`
+		logger.Infof("Execute: %s %v", command, args)
+		return testResponse, nil
+	}
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput:         executorFunc,
 		MockExecuteCommandWithCombinedOutput: executorFunc,
+		MockExecuteCommandWithTimeout:        executorFuncTimeout,
 	}
 
 	storeName := "myobject"