Repository: rook/rook
Pull request #13448 (2023-12-19 19:23:33)

csi: implement network fencing for cephFS (backport #13348)
This is an automatic backport of pull request #13348 done by [Mergify](https://mergify.com).


---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.com/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.com) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.com
</details>


Patch:
diff --git a/pkg/operator/ceph/cluster/watcher.go b/pkg/operator/ceph/cluster/watcher.go
index da55e20ce..25edc67d0 100644
--- a/pkg/operator/ceph/cluster/watcher.go
+++ b/pkg/operator/ceph/cluster/watcher.go
@@ -201,47 +201,79 @@ func (c *clientCluster) fenceNode(ctx context.Context, node *corev1.Node, cluste
 	}
 	logger.Debugf("volumesInuse %s", volumesInuse)
 
-	rbdVolumesInUse := getCephVolumesInUse(cluster, volumesInuse)
-	if len(rbdVolumesInUse) == 0 {
-		logger.Debugf("no rbd volumes in use for out of service node %q", node.Name)
+	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)
 		return nil
 	}
 
-	logger.Infof("node %q require fencing, found rbd volumes in use", node.Name)
 	listPVs, err := c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return pkgerror.Wrapf(err, "failed to list PV")
 	}
 
-	rbdPVList := listRBDPV(listPVs, cluster, rbdVolumesInUse)
-	if len(rbdPVList) == 0 {
-		logger.Debug("No rbd PVs found on the node")
-		return nil
-	}
+	if len(rbdVolumesInUse) != 0 {
+		rbdPVList := listRBDPV(listPVs, cluster, rbdVolumesInUse)
+		if len(rbdPVList) == 0 {
+			logger.Debug("No rbd PVs found on the node")
+		} else {
+			logger.Infof("node %q require fencing, found rbd 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.")
+			}
 
-	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 rbdPVList {
+				err = c.fenceRbdImage(ctx, node, cluster, clusterInfo, rbdPVList[i])
+				// We only need to create the network fence for any one of rbd pv.
+				if err == nil {
+					break
+				}
+
+				if i == len(rbdPVList)-1 {
+					return pkgerror.Wrapf(err, "failed to fence rbd volumes")
+				}
+				logger.Errorf("failed to fence rbd volumes %q, trying next rbd volume", rbdPVList[i].Name)
+			}
+		}
 	}
 
-	for i := range rbdPVList {
-		err = c.fenceRbdImage(ctx, node, cluster, clusterInfo, rbdPVList[i])
-		// We only need to create the network fence for any one of rbd pv.
-		if err == nil {
-			break
+	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
+			}
 
-		if i == len(rbdPVList)-1 {
-			return pkgerror.Wrapf(err, "failed to fence rbd volumes")
+			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)
 		}
-		logger.Errorf("failed to fence rbd volumes %q, trying next rbd volume", rbdPVList[i].Name)
+
 	}
 
 	return nil
 }
 
-func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.UniqueVolumeName) []string {
-	var rbdVolumesInUse []string
+func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.UniqueVolumeName) ([]string, []string) {
+	var rbdVolumesInUse, cephFSVolumeInUse []string
 
 	for _, volume := range volumesInUse {
 		splitVolumeInUseBased := trimeVolumeInUse(volume)
@@ -250,8 +282,13 @@ 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
+
+	return rbdVolumesInUse, cephFSVolumeInUse
 }
 
 func trimeVolumeInUse(volume corev1.UniqueVolumeName) []string {
@@ -290,6 +327,36 @@ 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" || pv.Spec.CSI.VolumeAttributes["pool"] == "" {
+				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 {
@@ -330,6 +397,80 @@ 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 fmt.Errorf("failed to get ceph status for check active mds. %v", err)
+	}
+
+	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 || ips == nil {
+		return fmt.Errorf("failed to unmarshal cephfs mds  output. %v", err)
+	}
+
+	err = c.createNetworkFence(ctx, cephFSPV, node, cluster, ips)
+	if err != nil {
+		return fmt.Errorf("failed to create network fence for node %q. %v", node.Name, err)
+	}
+
+	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 {
@@ -352,9 +493,10 @@ func rbdStatusUnMarshal(output []byte) ([]string, error) {
 }
 
 func concatenateWatcherIp(address string) string {
-	// address is in format `10.63.0.5:0/1254753579`
-	// 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"
+	// address is in format `10.63.0.5:0/1254753579` for rbd and
+	// in the format '10.244.0.12:0' for cephfs
+	// 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
 }
 
diff --git a/pkg/operator/ceph/csi/secrets.go b/pkg/operator/ceph/csi/secrets.go
index 324c26a70..5f11dd2e1 100644
--- a/pkg/operator/ceph/csi/secrets.go
+++ b/pkg/operator/ceph/csi/secrets.go
@@ -105,9 +105,10 @@ func cephCSIKeyringCephFSNodeCaps() []string {
 
 func cephCSIKeyringCephFSProvisionerCaps() []string {
 	return []string{
-		"mon", "allow r",
+		"mon", "allow r, allow command 'osd blocklist'",
 		"mgr", "allow rw",
 		"osd", "allow rw tag cephfs metadata=*",
+		"mds", "allow *", // TODO: replace '*' with required permissions
 	}
 }
 


Test patch:
diff --git a/pkg/operator/ceph/cluster/watcher_test.go b/pkg/operator/ceph/cluster/watcher_test.go
index 19d659437..d37cb84c1 100644
--- a/pkg/operator/ceph/cluster/watcher_test.go
+++ b/pkg/operator/ceph/cluster/watcher_test.go
@@ -175,9 +175,11 @@ func TestHandleNodeFailure(t *testing.T) {
 		switch {
 		case command == "rbd" && args[0] == "status":
 			return `{"watchers":[{"address":"192.168.39.137:0/3762982934","client":4307,"cookie":18446462598732840961}]}`, nil
+		case command == "ceph" && args[0] == "tell":
+			return `{"watchers":[{"id":5201,"entity":[{"addr": [{"addr": "10.244.0.12:0", "nonce":3247243972}]}]]}`, nil
 
 		}
-		return "", errors.Errorf("unexpected rbd command %q", args)
+		return "", errors.Errorf("unexpected rbd/ceph command %q", args)
 	}
 
 	node := &corev1.Node{
@@ -201,6 +203,7 @@ 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",
 			},
 		},
 	}
@@ -228,7 +231,7 @@ func TestHandleNodeFailure(t *testing.T) {
 		},
 	}
 
-	staticPV := &corev1.PersistentVolume{
+	staticRbdPV := &corev1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b841cb",
 			Annotations: map[string]string{
@@ -248,6 +251,26 @@ 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{},
+				},
+			},
+		},
+	}
+
 	pvNotProvisionByCSI := &corev1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "pvc-58469d41-f6c0-4720-b23a-0a0826b841cc",
@@ -302,30 +325,48 @@ func TestHandleNodeFailure(t *testing.T) {
 	err = c.client.Get(ctx, types.NamespacedName{Name: node.Name, Namespace: cephCluster.Namespace}, networkFence)
 	assert.NoError(t, err)
 
-	// For static pv
-	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, staticPV, metav1.CreateOptions{})
+	// For static rbd pv
+	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, staticRbdPV, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
 	pvList, err := c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	assert.NoError(t, err)
 
-	volumeInUse := getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
-	rbdPVList := listRBDPV(pvList, cephCluster, volumeInUse)
-	assert.Equal(t, len(rbdPVList), 1) // it will be equal to once since we have one pv provisioned by csi named `PV`
+	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`
 
 	err = c.handleNodeFailure(ctx, cephCluster, node)
 	assert.NoError(t, err)
 
-	// For static pv
+	// 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), 0)
+
+	err = c.handleNodeFailure(ctx, cephCluster, node)
+	assert.NoError(t, err)
+
+	// For pv not provisioned by CSI
 	_, err = c.context.Clientset.CoreV1().PersistentVolumes().Create(ctx, pvNotProvisionByCSI, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
 	pvList, err = c.context.Clientset.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	assert.NoError(t, err)
 
-	volumeInUse = getCephVolumesInUse(cephCluster, node.Status.VolumesInUse)
-	rbdPVList = listRBDPV(pvList, cephCluster, volumeInUse)
-	assert.Equal(t, len(rbdPVList), 1) // it will be equal to once since we have one pv provisioned by csi named `PV`
+	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`
 
 	err = c.handleNodeFailure(ctx, cephCluster, node)
 	assert.NoError(t, err)
@@ -345,6 +386,8 @@ 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])
@@ -355,9 +398,21 @@ 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")
 
-	trimVolInUse := getCephVolumesInUse(cephCluster, volInUse)
-	expected := []string{"0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4002", "0001-0009-rook-ceph-0000000000000002-24862838-240d-4215-9183-abfc0e9e4003"}
-	assert.Equal(t, expected, trimVolInUse)
+	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)
+
+	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) {
diff --git a/pkg/operator/ceph/csi/secrets_test.go b/pkg/operator/ceph/csi/secrets_test.go
index c7c908a94..d76b9db2a 100644
--- a/pkg/operator/ceph/csi/secrets_test.go
+++ b/pkg/operator/ceph/csi/secrets_test.go
@@ -39,5 +39,5 @@ func TestCephCSIKeyringCephFSNodeCaps(t *testing.T) {
 
 func TestCephCSIKeyringCephFSProvisionerCaps(t *testing.T) {
 	caps := cephCSIKeyringCephFSProvisionerCaps()
-	assert.Equal(t, caps, []string{"mon", "allow r", "mgr", "allow rw", "osd", "allow rw tag cephfs metadata=*"})
+	assert.Equal(t, caps, []string{"mon", "allow r, allow command 'osd blocklist'", "mgr", "allow rw", "osd", "allow rw tag cephfs metadata=*", "mds", "allow *"})
 }