Merge pull request #1532 from mowangdk/feature/support_bmcpfsid_fsetid_combination

Support fsetid static attach

Author: mowangdk

pkg/bmcpfs/bmcpfs.go

@@ -42,6 +42,8 @@ const (
 	// network types of CPFS mount targets
 	networkTypeVPC = "vpc"
 	networkTypeVSC = "vsc"
+
+	volumeHandleDelimiter = "+"
 )
 
 type Driver struct {

pkg/bmcpfs/controllerserver.go

@@ -43,6 +43,7 @@ type controllerServer struct {
 	vscManager     *internal.PrimaryVscManagerWithCache
 	attachDetacher internal.CPFSAttachDetacher
 	nasClient      *nasclient.Client
+	skipDetach     bool
 }
 
 func newControllerServer(region string) (*controllerServer, error) {
@@ -51,6 +52,11 @@ func newControllerServer(region string) (*controllerServer, error) {
 		return nil, err
 	}
 
+	skipDetach := false
+	if skipDetachVal, _ := strconv.ParseBool(os.Getenv("SKIP_BMCPFS_DETACH")); skipDetachVal {
+		skipDetach = skipDetachVal
+	}
+
 	nasClient, err := cloud.NewNasClientV2(region)
 	if err != nil {
 		return nil, err
@@ -59,6 +65,7 @@ func newControllerServer(region string) (*controllerServer, error) {
 		vscManager:     internal.NewPrimaryVscManagerWithCache(efloClient),
 		attachDetacher: internal.NewCPFSAttachDetacher(nasClient),
 		nasClient:      nasClient,
+		skipDetach:     skipDetach,
 	}, nil
 }
 
@@ -85,11 +92,12 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 		}, nil
 	}
 
+	cpfsID, _ := parseVolumeHandle(req.VolumeId)
 	// Get VscMountTarget of filesystem
 	mt := req.VolumeContext[_vscMountTarget]
 	if mt == "" {
 		var err error
-		mt, err = getMountTarget(cs.nasClient, req.VolumeId, networkTypeVSC)
+		mt, err = getMountTarget(cs.nasClient, cpfsID, networkTypeVSC)
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to get VscMountTarget: %v", err)
 		}
@@ -107,7 +115,7 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 	klog.Info("Use VSC MountTarget for lingjun node", "nodeId", req.NodeId, "vscId", vscId)
 
 	// Attach CPFS to VSC
-	err = cs.attachDetacher.Attach(ctx, req.VolumeId, vscId)
+	err = cs.attachDetacher.Attach(ctx, cpfsID, vscId)
 	if err != nil {
 		if autoSwitch, _ := strconv.ParseBool(req.VolumeContext[_mpAutoSwitch]); autoSwitch && internal.IsAttachNotSupportedError(err) {
 			if req.VolumeContext[_vpcMountTarget] == "" {
@@ -138,10 +146,9 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (
 	*csi.ControllerUnpublishVolumeResponse, error,
 ) {
-	if !strings.HasPrefix(req.NodeId, LingjunNodeIDPrefix) {
+	if !strings.HasPrefix(req.NodeId, LingjunNodeIDPrefix) || cs.skipDetach {
 		return &csi.ControllerUnpublishVolumeResponse{}, nil
 	}
-
 	// Create Primary vsc for Lingjun node
 	lingjunInstanceId := strings.TrimPrefix(req.NodeId, LingjunNodeIDPrefix)
 	if LingjunNodeIDPrefix == "" {
@@ -155,6 +162,7 @@ func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *
 		klog.InfoS("ControllerUnpublishVolume: skip detaching cpfs from vsc as vsc not found", "node", req.NodeId)
 		return &csi.ControllerUnpublishVolumeResponse{}, nil
 	}
+	// If `req.VolumeId` is a combination of `cpfsID` and `fsetID`, Detach will trigger an error.
 	err = cs.attachDetacher.Detach(ctx, req.VolumeId, vsc.VscID)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
@@ -209,6 +217,14 @@ func newEfloClient(region string) (*efloclient.Client, error) {
 	return efloclient.NewClient(config)
 }
 
+func parseVolumeHandle(volumeHandle string) (string, string) {
+	parts := strings.Split(volumeHandle, volumeHandleDelimiter)
+	if len(parts) == 2 {
+		return parts[0], parts[1]
+	}
+	return parts[0], ""
+}
+
 func getMountTarget(client *nasclient.Client, fsId, networkType string) (string, error) {
 	resp, err := client.DescribeFileSystems(&nasclient.DescribeFileSystemsRequest{
 		FileSystemId: &fsId,

pkg/bmcpfs/controllerserver_test.go

@@ -0,0 +1,60 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package bmcpfs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestParseVolumeHandle tests the parseVolumeHandle function with various inputs
+func TestParseVolumeHandle(t *testing.T) {
+	tests := []struct {
+		name           string
+		volumeHandle   string
+		expectedFsId   string
+		expectedFsetId string
+	}{
+		{
+			name:           "single part",
+			volumeHandle:   "fs-12345",
+			expectedFsId:   "fs-12345",
+			expectedFsetId: "",
+		},
+		{
+			name:           "two parts",
+			volumeHandle:   "fs-12345+fset-67890",
+			expectedFsId:   "fs-12345",
+			expectedFsetId: "fset-67890",
+		},
+		{
+			name:           "multiple delimiters",
+			volumeHandle:   "fs-12345+fset-67890+extra",
+			expectedFsId:   "fs-12345",
+			expectedFsetId: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fsId, fsetId := parseVolumeHandle(tt.volumeHandle)
+			assert.Equal(t, tt.expectedFsId, fsId)
+			assert.Equal(t, tt.expectedFsetId, fsetId)
+		})
+	}
+}