rgw: skip deleting root pool when there are multiple object stores

pkg/ceph/rgw/objectstore.go

@@ -31,7 +31,7 @@ const (
 
 var (
 	metadataPools = []string{
-		rootPool,
+		// .rgw.root (rootPool) is appended to this slice where needed
 		"rgw.control",
 		"rgw.meta",
 		"rgw.log",
@@ -64,12 +64,23 @@ func CreateObjectStore(context *Context, metadataSpec, dataSpec model.Pool, serv
 }
 
 func DeleteObjectStore(context *Context) error {
-	err := deleteRealm(context)
+	stores, err := GetObjectStores(context)
+	if err != nil {
+		return fmt.Errorf("failed to detect object stores during deletion. %+v", err)
+	}
+	logger.Infof("Found stores %v when deleting store %s", stores, context.Name)
+
+	err = deleteRealm(context)
 	if err != nil {
 		return fmt.Errorf("failed to delete realm. %+v", err)
 	}
 
-	err = deletePools(context)
+	lastStore := false
+	if len(stores) == 1 && stores[0] == context.Name {
+		lastStore = true
+	}
+
+	err = deletePools(context, lastStore)
 	if err != nil {
 		return fmt.Errorf("failed to delete object store pools. %+v", err)
 	}
@@ -195,8 +206,12 @@ func GetObjectStores(context *Context) ([]string, error) {
 	return r.Realms, nil
 }
 
-func deletePools(context *Context) error {
+func deletePools(context *Context, lastStore bool) error {
 	pools := append(metadataPools, dataPools...)
+	if lastStore {
+		pools = append(pools, rootPool)
+	}
+
 	for _, pool := range pools {
 		name := poolName(context.Name, pool)
 		if err := ceph.DeletePool(context.context, context.ClusterName, name); err != nil {
@@ -208,7 +223,7 @@ func deletePools(context *Context) error {
 }
 
 func createPools(context *Context, metadataSpec, dataSpec model.Pool) error {
-	if err := createSimilarPools(context, metadataPools, metadataSpec); err != nil {
+	if err := createSimilarPools(context, append(metadataPools, rootPool), metadataSpec); err != nil {
 		return fmt.Errorf("failed to create metadata pools. %+v", err)
 	}
 

pkg/ceph/rgw/objectstore_test.go

@@ -66,12 +66,18 @@ func TestCreateRealm(t *testing.T) {
 }
 
 func TestDeleteStore(t *testing.T) {
+	deleteStore(t, "myobj", `"mystore","myobj"`, false)
+	deleteStore(t, "myobj", `"myobj"`, true)
+}
+
+func deleteStore(t *testing.T, name string, existingStores string, expectedDeleteRootPool bool) {
 	realmDeleted := false
 	zoneDeleted := false
 	zoneGroupDeleted := false
 	poolsDeleted := 0
 	rulesDeleted := 0
 	executor := &exectest.MockExecutor{}
+	deletedRootPool := false
 	executor.MockExecuteCommandWithOutputFile = func(debug bool, actionName, command, outputFile string, args ...string) (string, error) {
 		//logger.Infof("command: %s %v", command, args)
 		if args[0] == "osd" {
@@ -81,6 +87,9 @@ func TestDeleteStore(t *testing.T) {
 				}
 				if args[2] == "delete" {
 					poolsDeleted++
+					if args[3] == rootPool {
+						deletedRootPool = true
+					}
 					return "", nil
 				}
 			}
@@ -96,9 +105,13 @@ func TestDeleteStore(t *testing.T) {
 	executor.MockExecuteCommandWithCombinedOutput = func(debug bool, actionName, command string, args ...string) (string, error) {
 		//logger.Infof("Command: %s %v", command, args)
 		if args[0] == "realm" {
-			assert.Equal(t, "delete", args[1])
-			realmDeleted = true
-			return "", nil
+			if args[1] == "delete" {
+				realmDeleted = true
+				return "", nil
+			}
+			if args[1] == "list" {
+				return fmt.Sprintf(`{"realms":[%s]}`, existingStores), nil
+			}
 		}
 		if args[0] == "zonegroup" {
 			assert.Equal(t, "delete", args[1])
@@ -114,11 +127,17 @@ func TestDeleteStore(t *testing.T) {
 	}
 	context := &Context{context: &clusterd.Context{Executor: executor}, Name: "myobj", ClusterName: "ns"}
 
+	// Delete an object store
 	err := DeleteObjectStore(context)
 	assert.Nil(t, err)
-	assert.Equal(t, 6, poolsDeleted)
-	assert.Equal(t, 6, rulesDeleted)
+	expectedPoolsDeleted := 5
+	if expectedDeleteRootPool {
+		expectedPoolsDeleted++
+	}
+	assert.Equal(t, expectedPoolsDeleted, poolsDeleted)
+	assert.Equal(t, expectedPoolsDeleted, rulesDeleted)
 	assert.True(t, realmDeleted)
 	assert.True(t, zoneGroupDeleted)
 	assert.True(t, zoneDeleted)
+	assert.Equal(t, expectedDeleteRootPool, deletedRootPool)
 }