Merge pull request #13801 from subhamkrai/disable-cephfs-fencing

csi: disable cephFs network Fencing temporarily

pkg/operator/ceph/cluster/watcher.go

@@ -60,8 +60,7 @@ var (
 
 // drivers that supports fencing, used in naming networkFence object
 const (
-	rbdDriver    = "rbd"
-	cephfsDriver = "cephfs"
+	rbdDriver = "rbd"
 )
 
 func newClientCluster(client client.Client, namespace string, context *clusterd.Context) *clientCluster {
@@ -201,11 +200,6 @@ func (c *clientCluster) handleNodeFailure(ctx context.Context, cluster *cephv1.C
 		return pkgerror.Wrapf(err, "failed to delete rbd network fence for node %q.", node.Name)
 	}
 
-	err = c.unfenceAndDeleteNetworkFence(ctx, *node, cluster, cephfsDriver)
-	if err != nil {
-		return pkgerror.Wrapf(err, "failed to delete cephFS network fence for node %q.", node.Name)
-	}
-
 	return nil
 }
 
@@ -217,9 +211,9 @@ func (c *clientCluster) fenceNode(ctx context.Context, node *corev1.Node, cluste
 	}
 	logger.Debugf("volumesInuse %s", volumesInuse)
 
-	rbdVolumesInUse, cephFSVolumeInUse := getCephVolumesInUse(cluster, volumesInuse)
-	if len(rbdVolumesInUse) == 0 && len(cephFSVolumeInUse) == 0 {
-		logger.Debugf("no rbd or cephFS volumes in use for out of service node %q", node.Name)
+	rbdVolumesInUse := getCephVolumesInUse(cluster, volumesInuse)
+	if len(rbdVolumesInUse) == 0 {
+		logger.Debugf("no rbd  volumes in use for out of service node %q", node.Name)
 		return nil
 	}
 
@@ -258,46 +252,11 @@ func (c *clientCluster) fenceNode(ctx context.Context, node *corev1.Node, cluste
 		}
 	}
 
-	if len(cephFSVolumeInUse) != 0 {
-		cephFSVolumeInUseMap := make(map[string]struct{})
-		for _, vol := range cephFSVolumeInUse {
-			cephFSVolumeInUseMap[vol] = struct{}{}
-		}
-		cephFSPVList := listRWOCephFSPV(listPVs, cluster, cephFSVolumeInUseMap)
-		if len(cephFSPVList) == 0 {
-			logger.Debug("No cephFS PVs found on the node")
-			return nil
-		}
-		logger.Infof("node %q require fencing, found cephFS volumes in use", node.Name)
-		clusterInfo, _, _, err := opcontroller.LoadClusterInfo(c.context, ctx, cluster.Namespace, &cluster.Spec)
-		if err != nil {
-			return pkgerror.Wrapf(err, "Failed to load cluster info.")
-		}
-
-		for i := range cephFSPVList {
-			err = c.fenceCephFSVolume(ctx, node, cluster, clusterInfo, cephFSPVList[i])
-			// We only need to create the network fence for any one of cephFS pv.
-			if err == nil {
-				break
-			}
-
-			// continue to fence next rbd volume if active client not found
-			if stderrors.Is(err, errActiveClientNotFound) {
-				continue
-			}
-			if i == len(cephFSPVList)-1 {
-				return pkgerror.Wrapf(err, "failed to fence cephFS volumes")
-			}
-			logger.Errorf("failed to fence cephFS volumes %q, trying next cephFS volume", cephFSPVList[i].Name)
-		}
-
-	}
-
 	return nil
 }
 
-func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.UniqueVolumeName) ([]string, []string) {
-	var rbdVolumesInUse, cephFSVolumeInUse []string
+func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.UniqueVolumeName) []string {
+	var rbdVolumesInUse []string
 
 	for _, volume := range volumesInUse {
 		splitVolumeInUseBased := trimeVolumeInUse(volume)
@@ -306,13 +265,9 @@ func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.Uniq
 		if len(splitVolumeInUseBased) == 2 && splitVolumeInUseBased[0] == fmt.Sprintf("%s.rbd.csi.ceph.com", cluster.Namespace) {
 			rbdVolumesInUse = append(rbdVolumesInUse, splitVolumeInUseBased[1])
 		}
-
-		if len(splitVolumeInUseBased) == 2 && splitVolumeInUseBased[0] == fmt.Sprintf("%s.cephfs.csi.ceph.com", cluster.Namespace) {
-			cephFSVolumeInUse = append(cephFSVolumeInUse, splitVolumeInUseBased[1])
-		}
 	}
 
-	return rbdVolumesInUse, cephFSVolumeInUse
+	return rbdVolumesInUse
 }
 
 func trimeVolumeInUse(volume corev1.UniqueVolumeName) []string {
@@ -351,36 +306,6 @@ func listRBDPV(listPVs *corev1.PersistentVolumeList, cluster *cephv1.CephCluster
 	return listRbdPV
 }
 
-func listRWOCephFSPV(listPVs *corev1.PersistentVolumeList, cluster *cephv1.CephCluster, cephFSVolumesInUse map[string]struct{}) []corev1.PersistentVolume {
-	var listCephFSPV []corev1.PersistentVolume
-
-	for _, pv := range listPVs.Items {
-		// Skip if pv is not provisioned by CSI
-		if pv.Spec.CSI == nil {
-			logger.Debugf("pv %q is not provisioned by CSI", pv.Name)
-			continue
-		}
-
-		if pv.Spec.CSI.Driver == fmt.Sprintf("%s.cephfs.csi.ceph.com", cluster.Namespace) {
-			// Ignore PVs that support multinode access (RWX, ROX), since they can be mounted on multiple nodes.
-			if pvSupportsMultiNodeAccess(pv.Spec.AccessModes) {
-				continue
-			}
-
-			if pv.Spec.CSI.VolumeAttributes["staticVolume"] == "true" {
-				logger.Debugf("skipping, static pv %q", pv.Name)
-				continue
-			}
-			// Check if the volume is in use
-			if _, exists := cephFSVolumesInUse[pv.Spec.CSI.VolumeHandle]; exists {
-				listCephFSPV = append(listCephFSPV, pv)
-			}
-		}
-
-	}
-	return listCephFSPV
-}
-
 // pvSupportsMultiNodeAccess returns true if the PV access modes contain ReadWriteMany or ReadOnlyMany.
 func pvSupportsMultiNodeAccess(accessModes []corev1.PersistentVolumeAccessMode) bool {
 	for _, accessMode := range accessModes {
@@ -423,85 +348,6 @@ func (c *clientCluster) fenceRbdImage(
 	return nil
 }
 
-func (c *clientCluster) fenceCephFSVolume(
-	ctx context.Context, node *corev1.Node, cluster *cephv1.CephCluster,
-	clusterInfo *cephclient.ClusterInfo, cephFSPV corev1.PersistentVolume) error {
-
-	logger.Infof("fencing cephfs volume %q on node %q", cephFSPV.Name, node.Name)
-
-	status, err := cephclient.StatusWithUser(c.context, clusterInfo)
-	if err != nil {
-		return pkgerror.Wrapf(err, "failed to get ceph status for check active mds")
-	}
-
-	var activeMDS string
-	for _, fsRank := range status.Fsmap.ByRank {
-		if fsRank.Status == "up:active" {
-			activeMDS = fsRank.Name
-		}
-	}
-
-	args := []string{"tell", fmt.Sprintf("mds.%s", activeMDS), "client", "ls", "--format", "json"}
-	cmd := cephclient.NewCephCommand(c.context, clusterInfo, args)
-	cmd.JsonOutput = true
-
-	buf, err := cmd.Run()
-	if err != nil {
-		return fmt.Errorf("failed to list watchers for cephfs pool/subvoumeName %s/%s. %v", cephFSPV.Spec.CSI.VolumeAttributes["pool"], cephFSPV.Spec.CSI.VolumeAttributes["subvolumeName"], err)
-	}
-	ips, err := cephFSMDSClientMarshal(buf, cephFSPV)
-	if err != nil {
-		return pkgerror.Wrapf(err, "failed to unmarshal cephfs mds output")
-	}
-
-	if len(ips) == 0 {
-		logger.Infof("no active mds clients found for cephfs volume %q", cephFSPV.Name)
-		return errActiveClientNotFound
-	}
-
-	err = c.createNetworkFence(ctx, cephFSPV, node, cluster, ips, cephfsDriver)
-	if err != nil {
-		return pkgerror.Wrapf(err, "failed to create network fence for node %q", node.Name)
-	}
-
-	return nil
-}
-
-func cephFSMDSClientMarshal(output []byte, cephFSPV corev1.PersistentVolume) ([]string, error) {
-	type entity struct {
-		Addr struct {
-			Addr  string `json:"addr"`
-			Nonce int    `json:"nonce"`
-		} `json:"addr"`
-	}
-
-	type clientMetadata struct {
-		Root string `json:"root"`
-	}
-
-	type cephFSData struct {
-		Entity         entity         `json:"entity"`
-		ClientMetadata clientMetadata `json:"client_metadata"`
-	}
-
-	var data []cephFSData
-	err := json.Unmarshal([]byte(output), &data)
-	if err != nil {
-		return []string{}, pkgerror.Wrapf(err, "failed to unmarshal cephFS data output")
-	}
-
-	watcherIPlist := []string{}
-	for _, d := range data {
-		if cephFSPV.Spec.CSI.VolumeAttributes["subvolumePath"] == d.ClientMetadata.Root {
-			logger.Infof("cephfs mds client ips to fence %v", d.Entity.Addr)
-			watcherIP := concatenateWatcherIp(d.Entity.Addr.Addr)
-			watcherIPlist = append(watcherIPlist, watcherIP)
-		}
-	}
-
-	return watcherIPlist, nil
-}
-
 func rbdStatusUnMarshal(output []byte) ([]string, error) {
 	type rbdStatus struct {
 		Watchers []struct {
@@ -524,8 +370,7 @@ func rbdStatusUnMarshal(output []byte) ([]string, error) {
 }
 
 func concatenateWatcherIp(address string) string {
-	// address is in format `10.63.0.5:0/1254753579` for rbd and
-	// in the format '10.244.0.12:0' for cephfs
+	// address is in format `10.63.0.5:0/1254753579` for rbd
 	// split with separation ':0' to remove nounce and concatenating `/32` to define a network with only one IP address
 	watcherIP := strings.Split(address, ":0")[0] + "/32"
 	return watcherIP
@@ -604,8 +449,8 @@ func (c *clientCluster) unfenceAndDeleteNetworkFence(ctx context.Context, node c
 			return false, err
 		}
 
-		if networkFence.Spec.FenceState != addonsv1alpha1.Unfenced {
-			logger.Infof("waiting for network fence CR %s to get in %s state before deletion", networkFence.Name, addonsv1alpha1.Unfenced)
+		if networkFence.Status.Message != addonsv1alpha1.UnFenceOperationSuccessfulMessage {
+			logger.Infof("waiting for network fence CR %q status to get result %q", networkFence.Name, addonsv1alpha1.UnFenceOperationSuccessfulMessage)
 			return false, err
 		}
 

pkg/operator/ceph/cluster/watcher_test.go

@@ -206,7 +206,6 @@ func TestHandleNodeFailure(t *testing.T) {
 			},
 			VolumesInUse: []corev1.UniqueVolumeName{
 				"kubernetes.io/csi/rook-ceph.rbd.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4002",
-				"kubernetes.io/csi/rook-ceph.cephfs.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001",
 			},
 		},
 	}
@@ -234,30 +233,6 @@ func TestHandleNodeFailure(t *testing.T) {
 		},
 	}
 
-	cephfsPV := &corev1.PersistentVolume{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b842ca",
-			Annotations: map[string]string{
-				"pv.kubernetes.io/provisioned-by":                            fmt.Sprintf("%s.cephfs.csi.ceph.com", ns),
-				"volume.kubernetes.io/provisioner-deletion-secret-name":      "rook-csi-cephfs-provisioner",
-				"volume.kubernetes.io/provisioner-deletion-secret-namespace": ns,
-			},
-		},
-		Spec: corev1.PersistentVolumeSpec{
-			PersistentVolumeSource: corev1.PersistentVolumeSource{
-				CSI: &corev1.CSIPersistentVolumeSource{
-					Driver:       fmt.Sprintf("%s.cephfs.csi.ceph.com", ns),
-					VolumeHandle: "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001",
-					VolumeAttributes: map[string]string{
-						"fsName":        "myfs",
-						"subvolumePath": "/volumes/csi/csi-vol-58469d41-f6c0-4720-b23a-0a0826b842ca",
-						"subvolumeName": "csi-vol-58469d41-f6c0-4720-b23a-0a0826b842ca",
-					},
-				},
-			},
-		},
-	}
-
 	staticRbdPV := &corev1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b841cb",
@@ -278,28 +253,6 @@ func TestHandleNodeFailure(t *testing.T) {
 		},
 	}
 
-	staticCephfsPV := &corev1.PersistentVolume{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b842cb",
-			Annotations: map[string]string{
-				"pv.kubernetes.io/provisioned-by":                            fmt.Sprintf("%s.cephfs.csi.ceph.com", ns),
-				"volume.kubernetes.io/provisioner-deletion-secret-name":      "rook-csi-cephfs-provisioner",
-				"volume.kubernetes.io/provisioner-deletion-secret-namespace": ns,
-			},
-		},
-		Spec: corev1.PersistentVolumeSpec{
-			PersistentVolumeSource: corev1.PersistentVolumeSource{
-				CSI: &corev1.CSIPersistentVolumeSource{
-					Driver:       fmt.Sprintf("%s.cephfs.csi.ceph.com", ns),
-					VolumeHandle: "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001",
-					VolumeAttributes: map[string]string{
-						"staticVolume": "true",
-					},
-				},
-			},
-		},
-	}
-
 	pvNotProvisionByCSI := &corev1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b841cc",
@@ -343,9 +296,6 @@ func TestHandleNodeFailure(t *testing.T) {
 	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, rbdPV, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
-	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, cephfsPV, metav1.CreateOptions{})
-	assert.NoError(t, err)
-
 	_, err = c.context.ApiExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Create(ctx, &v1.CustomResourceDefinition{ObjectMeta: metav1.ObjectMeta{Name: "networkfences.csiaddons.openshift.io"}}, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
@@ -357,26 +307,19 @@ func TestHandleNodeFailure(t *testing.T) {
 	err = c.client.Get(ctx, types.NamespacedName{Name: fenceResourceName(node.Name, rbdDriver, ns)}, networkFenceRbd)
 	assert.NoError(t, err)
 
-	networkFenceCephFs := &addonsv1alpha1.NetworkFence{}
-	err = c.client.Get(ctx, types.NamespacedName{Name: fenceResourceName(node.Name, cephfsDriver, ns)}, networkFenceCephFs)
-	assert.NoError(t, err)
-
 	networkFences := &addonsv1alpha1.NetworkFenceList{}
 	err = c.client.List(ctx, networkFences)
 	assert.NoError(t, err)
-	var rbdCount, cephFsCount int
+	var rbdCount int
 
 	for _, fence := range networkFences.Items {
 		// Check if the resource is in the desired namespace
 		if strings.Contains(fence.Name, rbdDriver) {
 			rbdCount++
-		} else if strings.Contains(fence.Name, cephfsDriver) {
-			cephFsCount++
 		}
 	}
 
 	assert.Equal(t, 1, rbdCount)
-	assert.Equal(t, 1, cephFsCount)
 
 	// For static rbd pv
 	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, staticRbdPV, metav1.CreateOptions{})
@@ -385,28 +328,13 @@ func TestHandleNodeFailure(t *testing.T) {
 	pvList, err := c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	assert.NoError(t, err)
 
-	rbdVolumesInUse, _ := getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
+	rbdVolumesInUse := getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
 	rbdPVList := listRBDPV(pvList, cephCluster, rbdVolumesInUse)
 	assert.Equal(t, len(rbdPVList), 1) // it will be equal to one since we have one pv provisioned by csi named `rbdPV`
 
 	err = c.handleNodeFailure(ctx, cephCluster, node)
 	assert.NoError(t, err)
 
-	// For static cephfs pv
-	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, staticCephfsPV, metav1.CreateOptions{})
-	assert.NoError(t, err)
-
-	pvList, err = c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
-	assert.NoError(t, err)
-
-	_, cephFSVolumesInUse := getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
-	cephFSVolumesInUseMap := make(map[string]struct{})
-	for _, vol := range cephFSVolumesInUse {
-		cephFSVolumesInUseMap[vol] = struct{}{}
-	}
-	cephFSPVList := listRWOCephFSPV(pvList, cephCluster, cephFSVolumesInUseMap)
-	assert.Equal(t, len(cephFSPVList), 1) // it will be equal to one since we have one pv provisioned by csi named `cephfsPV`
-
 	err = c.handleNodeFailure(ctx, cephCluster, node)
 	assert.NoError(t, err)
 
@@ -417,7 +345,7 @@ func TestHandleNodeFailure(t *testing.T) {
 	pvList, err = c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	assert.NoError(t, err)
 
-	rbdVolumesInUse, _ = getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
+	rbdVolumesInUse = getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
 	rbdPVList = listRBDPV(pvList, cephCluster, rbdVolumesInUse)
 	assert.Equal(t, len(rbdPVList), 1) // it will be equal to one since we have one pv provisioned by csi named `PV`
 
@@ -427,15 +355,15 @@ func TestHandleNodeFailure(t *testing.T) {
 	// When out-of-service taint is removed
 	node.Spec.Taints = []corev1.Taint{}
 
+	networkFenceRbd.Status.Message = addonsv1alpha1.UnFenceOperationSuccessfulMessage
+	err = c.client.Update(ctx, networkFenceRbd)
+	assert.NoError(t, err)
+
 	err = c.handleNodeFailure(ctx, cephCluster, node)
 	assert.NoError(t, err)
 
 	err = c.client.Get(ctx, types.NamespacedName{Name: fenceResourceName(node.Name, rbdDriver, ns), Namespace: cephCluster.Namespace}, networkFenceRbd)
 	assert.Error(t, err, kerrors.IsNotFound(err))
-
-	err = c.client.Get(ctx, types.NamespacedName{Name: fenceResourceName(node.Name, cephfsDriver, ns), Namespace: cephCluster.Namespace}, networkFenceCephFs)
-	assert.Error(t, err, kerrors.IsNotFound(err))
-
 }
 
 func TestGetCephVolumesInUse(t *testing.T) {
@@ -443,8 +371,6 @@ func TestGetCephVolumesInUse(t *testing.T) {
 	volInUse := []corev1.UniqueVolumeName{
 		"kubernetes.io/csi/rook-ceph.rbd.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4002",
 		"kubernetes.io/csi/rook-ceph.rbd.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4003",
-		"kubernetes.io/csi/rook-ceph.cephfs.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001",
-		"kubernetes.io/csi/rook-ceph.cephfs.csi.ceph.com^0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4004",
 	}
 
 	splitVolInUse := trimeVolumeInUse(volInUse[0])
@@ -455,21 +381,11 @@ func TestGetCephVolumesInUse(t *testing.T) {
 	assert.Equal(t, splitVolInUse[0], "rook-ceph.rbd.csi.ceph.com")
 	assert.Equal(t, splitVolInUse[1], "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4003")
 
-	splitVolInUse = trimeVolumeInUse(volInUse[2])
-	assert.Equal(t, splitVolInUse[0], "rook-ceph.cephfs.csi.ceph.com")
-	assert.Equal(t, splitVolInUse[1], "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001")
-
-	splitVolInUse = trimeVolumeInUse(volInUse[3])
-	assert.Equal(t, splitVolInUse[0], "rook-ceph.cephfs.csi.ceph.com")
-	assert.Equal(t, splitVolInUse[1], "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4004")
-
-	trimRbdVolInUse, trimCephFSVolInUse := getCephVolumesInUse(cephCluster, volInUse)
+	trimRbdVolInUse := getCephVolumesInUse(cephCluster, volInUse)
 
 	expectedRbd := []string{"0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4002", "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4003"}
-	expectedCephfs := []string{"0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4001", "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4004"}
 
 	assert.Equal(t, expectedRbd, trimRbdVolInUse)
-	assert.Equal(t, expectedCephfs, trimCephFSVolInUse)
 }
 
 func TestRBDStatusUnMarshal(t *testing.T) {