From efe6ef4e451005b082318019dc22e75076e510d3 Mon Sep 17 00:00:00 2001 From: subhamkrai Date: Fri, 17 May 2024 12:04:17 +0530 Subject: [PATCH] csi: revert remove cephFs networkFence code temporarily This reverts commit bd4250f56e202daf9cf40ae3dd16382cb75663df. Signed-off-by: subhamkrai --- pkg/operator/ceph/cluster/watcher.go | 171 +++++++++++++++++++++- pkg/operator/ceph/cluster/watcher_test.go | 100 ++++++++++++- 2 files changed, 255 insertions(+), 16 deletions(-) diff --git a/pkg/operator/ceph/cluster/watcher.go b/pkg/operator/ceph/cluster/watcher.go index a79c71221..79aac4fe7 100644 --- a/pkg/operator/ceph/cluster/watcher.go +++ b/pkg/operator/ceph/cluster/watcher.go @@ -60,7 +60,8 @@ var ( // drivers that supports fencing, used in naming networkFence object const ( - rbdDriver = "rbd" + rbdDriver = "rbd" + cephfsDriver = "cephfs" ) func newClientCluster(client client.Client, namespace string, context *clusterd.Context) *clientCluster { @@ -210,6 +211,11 @@ 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 } @@ -221,9 +227,9 @@ 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 } @@ -262,11 +268,46 @@ 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 { - var rbdVolumesInUse []string +func getCephVolumesInUse(cluster *cephv1.CephCluster, volumesInUse []corev1.UniqueVolumeName) ([]string, []string) { + var rbdVolumesInUse, cephFSVolumeInUse []string for _, volume := range volumesInUse { splitVolumeInUseBased := trimeVolumeInUse(volume) @@ -275,9 +316,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 { @@ -316,6 +361,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" { + 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 { @@ -358,6 +433,85 @@ 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 { @@ -380,7 +534,8 @@ func rbdStatusUnMarshal(output []byte) ([]string, error) { } func concatenateWatcherIp(address string) string { - // address is in format `10.63.0.5:0/1254753579` for rbd + // 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/cluster/watcher_test.go b/pkg/operator/ceph/cluster/watcher_test.go index e3585ae81..29b6d60f8 100644 --- a/pkg/operator/ceph/cluster/watcher_test.go +++ b/pkg/operator/ceph/cluster/watcher_test.go @@ -206,6 +206,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", }, }, } @@ -233,6 +234,30 @@ 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", @@ -253,6 +278,28 @@ 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", @@ -296,6 +343,9 @@ 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) @@ -307,19 +357,26 @@ 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 int + var rbdCount, cephFsCount 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{}) @@ -328,13 +385,28 @@ 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) @@ -345,7 +417,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` @@ -355,15 +427,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) { @@ -371,6 +443,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]) @@ -381,11 +455,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") - trimRbdVolInUse := getCephVolumesInUse(cephCluster, volInUse) + 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) {