fix: Enhanced cephfs enable workflow

Signed-off-by: Utkarsh Bhatt <utkarsh.bhatt@canonical.com>

Author: UtkarshBhatthere

microceph/ceph/cephfs_volume.go

@@ -56,7 +56,25 @@ func ListCephFSVolumes() ([]Volume, error) {
 		response = append(response, Volume(volume.String()))
 	}
 
-	return response, nil
+	return parseListVolOutputJson(output)
+}
+
+func ListRemoteCephFSVolumes(remote string, local string) ([]Volume, error) {
+	if len(remote) == 0 || len(local) == 0 {
+		return nil, fmt.Errorf("both remote(%s) and local(%s) names must be provided", remote, local)
+	}
+
+	args := []string{"fs", "volume", "ls", "--format=json"}
+	cmd := appendRemoteClusterArgs(args, remote, local)
+
+	output, err := cephRun(cmd...)
+	if err != nil {
+		return nil, err
+	}
+
+	logger.Infof("FSVOL: listed %s as remote cluster (%s) cephfs volumes", output, remote)
+
+	return parseListVolOutputJson(output)
 }
 
 // CephFsSubvolumePathDeconstruct deconstructs a CephFS subvolume path string into its subvolume and subvolumegroup names.
@@ -182,3 +200,20 @@ func GetCephFSSubvolumePath(subvolumegroup string, subvolume string) string {
 
 	return retval
 }
+
+// ##### Helper Functions #####
+
+func parseListVolOutputJson(output string) ([]Volume, error) {
+	volumes := gjson.Get(output, "#.name")
+	response := make([]Volume, 0, len(volumes.Array()))
+	for _, volume := range volumes.Array() {
+		if len(volume.String()) == 0 {
+			continue
+		}
+
+		logger.Debugf("FSVOL: found %s as cephfs volume", volume.String())
+		response = append(response, Volume(volume.String()))
+	}
+
+	return response, nil
+}

microceph/ceph/manager.go

@@ -1,10 +1,13 @@
 package ceph
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
+	"strings"
 
 	"github.com/canonical/microceph/microceph/common"
+	"github.com/canonical/microceph/microceph/constants"
 	"github.com/canonical/microceph/microceph/logger"
 	"github.com/tidwall/gjson"
 )
@@ -50,11 +53,46 @@ func getActiveMgrs() ([]string, error) {
 }
 
 // Mgr Module Ops
-func enableMgrModule(module string) error {
-	cmd := []string{
-		"mgr", "module", "enable", module,
-	}
+
+// EnableMgrModule enabled a mgr module on specified ceph cluster and verifies if is comes up
+func EnableMgrModule(ctx context.Context, module string, remote string, local string) error {
+	args := []string{"mgr", "module", "enable", module}
+
+	cmd := appendRemoteClusterArgs(args, remote, local)
 
 	_, err := cephRun(cmd...)
-	return err
+	if err != nil {
+		logger.Errorf("Failed to enable remote cluster (%s) mgr module %s: %v", remote, module, err)
+		return err
+	}
+
+	return verifyMgrModuleEnabled(ctx, module, remote, local)
+}
+
+func verifyMgrModuleEnabled(ctx context.Context, module string, remote string, local string) error {
+	args := []string{"mgr", "module", "ls", "-f", "json"}
+
+	cmd := appendRemoteClusterArgs(args, remote, local)
+
+	output, err := cephRunContext(ctx, cmd...)
+	if err != nil {
+		logger.Errorf("Failed to list remote cluster (%s) mgr modules: %v", remote, err)
+		return err
+	}
+
+	result := gjson.Get(output, "enabled_modules")
+	if !result.Exists() || !result.Bool() {
+		return fmt.Errorf("mgr module %s not enabled", module)
+	}
+
+	logger.Debugf("Enabled Mgr Modules:\n%+v", result.Array())
+
+	for _, module := range result.Array() {
+		logger.Debugf("Mgr Module: %s", module.String())
+		if strings.Compare(module.String(), constants.MgrModuleMirroring) == 0 {
+			return nil
+		}
+	}
+
+	return fmt.Errorf("could not enable mgr module %s", module)
 }

microceph/ceph/rbd_mirror.go

@@ -3,18 +3,19 @@ package ceph
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/canonical/microceph/microceph/common"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
+	"github.com/canonical/microceph/microceph/common"
+
 	"github.com/Rican7/retry"
 	"github.com/Rican7/retry/backoff"
 	"github.com/Rican7/retry/strategy"
-	"github.com/canonical/microceph/microceph/logger"
 	"github.com/canonical/microceph/microceph/api/types"
 	"github.com/canonical/microceph/microceph/constants"
+	"github.com/canonical/microceph/microceph/logger"
 	"github.com/tidwall/gjson"
 )
 
@@ -680,7 +681,7 @@ func IsRemoteConfiguredForRbdMirror(remoteName string) bool {
 
 // appendRemoteClusterArgs appends the cluster and client arguments to ceph commands
 func appendRemoteClusterArgs(args []string, cluster string, client string) []string {
-	logger.Debugf("RBD Replication: old args are %v", args)
+	logger.Debugf("REP: old args are %v", args)
 	// check if appendage is needed
 	if len(cluster) == 0 && len(client) == 0 {
 		// return as is
@@ -697,7 +698,7 @@ func appendRemoteClusterArgs(args []string, cluster string, client string) []str
 		args = append(args, client)
 	}
 
-	logger.Debugf("RBD Replication: new args are %v", args)
+	logger.Debugf("REP: new args are %v", args)
 
 	// return modified args
 	return args

microceph/ceph/replication_cephfs.go

@@ -131,7 +131,22 @@ func (rh *CephfsReplicationHandler) EnableHandler(ctx context.Context, args ...a
 		return err
 	}
 
-	err = enableMgrModule(constants.MgrModuleMirroring)
+	dbRec, err := database.GetRemoteDb(ctx, st.ClusterState(), rh.Request.RemoteName)
+	if err != nil {
+		err := fmt.Errorf("remote (%s) does not exist: %w", rh.Request.RemoteName, err)
+		logger.Error(err.Error())
+		return err
+	}
+
+	logger.Infof("REPCFS: LocalName(%s) RemoteName(%s)", dbRec[0].LocalName, dbRec[0].Name)
+
+	err = verifyRemoteCephFSVolumeExists(rh.Request.Volume, dbRec[0].Name, dbRec[0].LocalName)
+	if err != nil {
+		logger.Errorf("Failed to verify if remote cephfs volume %s: %v", rh.Request.Volume, err)
+		return err
+	}
+
+	err = enableCephFSMgrModules(ctx, dbRec[0].Name, dbRec[0].LocalName)
 	if err != nil {
 		return err
 	}
@@ -237,7 +252,7 @@ func (rh *CephfsReplicationHandler) StatusHandler(ctx context.Context, args ...a
 
 			// cannot directly modify struct fields in a map, so retrieve, modify and reassign
 			for _, peer := range fs.Peers {
-				responsePeer, _ := response.Peers[peer.UUID]
+				responsePeer := response.Peers[peer.UUID]
 				responsePeer.Name = peer.Remote.ClusterName
 				response.Peers[peer.UUID] = responsePeer
 			}
@@ -279,9 +294,9 @@ func (rh *CephfsReplicationHandler) GetCephFSMirrorStatus(ctx context.Context) e
 		return fmt.Errorf("%s is not %s", rh.Request.RequestType, types.StatusReplicationRequest)
 	}
 
-	volumeId, peers := GetCephFsMirrorVolumeAndPeersId(rh)
-	if volumeId < 0 || len(peers) == 0 {
-		return fmt.Errorf("no CephFS volume (%d) or peers (%v) found for mirroring status", volumeId, peers)
+	volumeID, peers := GetCephFsMirrorVolumeAndPeersId(rh)
+	if volumeID < 0 || len(peers) == 0 {
+		return fmt.Errorf("no CephFS volume (%d) or peers (%v) found for mirroring status", volumeID, peers)
 	}
 
 	// TODO: (utkarshbhatthere):
@@ -297,7 +312,7 @@ func (rh *CephfsReplicationHandler) GetCephFSMirrorStatus(ctx context.Context) e
 	response := MirrorStatus{}
 	for _, peer := range peers {
 		// Get the mirror status for each peer
-		response[peer], err = GetCephFsMirrorPeerStatus(ctx, cephfsMirrorAdminSock, volumeId, peer)
+		response[peer], err = GetCephFsMirrorPeerStatus(ctx, cephfsMirrorAdminSock, volumeID, peer)
 		if err != nil {
 			return fmt.Errorf("failed to get CephFS mirror status for peer %s: %w", peer, err)
 		}
@@ -368,6 +383,40 @@ func verifyEnableRequestData(ctx context.Context, s interfaces.CephState, reques
 	return nil
 }
 
+func enableCephFSMgrModules(ctx context.Context, remote string, local string) error {
+	err := EnableMgrModule(ctx, constants.MgrModuleMirroring, "", "")
+	if err != nil {
+		err := fmt.Errorf("failed to enable mgr module %s on local cluster: %w", constants.MgrModuleMirroring, err)
+		logger.Error(err.Error())
+		return err
+	}
+
+	err = EnableMgrModule(ctx, constants.MgrModuleMirroring, remote, local)
+	if err != nil {
+		err := fmt.Errorf("failed to enable mgr module %s on remote cluster %s: %w", constants.MgrModuleMirroring, remote, err)
+		logger.Error(err.Error())
+		return err
+	}
+
+	return nil
+}
+
+func verifyRemoteCephFSVolumeExists(vol string, remote string, local string) error {
+	volumes, err := ListRemoteCephFSVolumes(remote, local)
+	if err != nil {
+		logger.Errorf("Failed to list remote cephfs volumes for remote %s: %v", remote, err)
+		return err
+	}
+
+	for _, volume := range volumes {
+		if strings.Compare(string(volume), vol) == 0 {
+			return nil
+		}
+	}
+
+	return fmt.Errorf("cephfs volume %s not found on remote %s", vol, remote)
+}
+
 func enableCephFSVolumeMirror(ctx context.Context, request types.CephfsReplicationRequest) error {
 	peerExists, err := cephFSSnapshotMirrorPeerExists(ctx, request.Volume, request.RemoteName)
 	if err != nil {