Support bmcpfs fileset dynamic provisioning

Author: mowangdk

.golangci.yaml

@@ -23,6 +23,11 @@ issues:
     - path: .*\.go 
       linters:
         - unused
+    # Temporary: suppress deprecated Endpoints API warning until dadi migrates to EndpointSlice.
+    - path: pkg/dadi/.*\.go
+      text: corev1\.Endpoints is deprecated
+      linters:
+        - staticcheck
 
   # Mode of the generated files analysis.
   #

_typos.toml

@@ -10,3 +10,5 @@ teh = "teh"
 ded = "ded"
 eles = "eles"
 TAGED = "TAGED"
+Entrie = "Entrie"
+DescribeFilesetsResponseBodyEntriesEntrie = "DescribeFilesetsResponseBodyEntriesEntrie"

pkg/bmcpfs/bmcpfs.go

@@ -30,13 +30,14 @@ const (
 	// keys in volume context or publish context
 	_vpcMountTarget = "vpcMountTarget"
 	_vscMountTarget = "vscMountTarget"
-	_vscId          = "vscId"
+	_vscID          = "vscId"
 	_networkType    = "networkType"
 	_path           = "path"
 	_mpAutoSwitch   = "mountpointAutoSwitch"
 
-	// prefix of node id
-	CommonNodeIDPrefix  = "common:"
+	// CommonNodeIDPrefix is the prefix for common node IDs
+	CommonNodeIDPrefix = "common:"
+	// LingjunNodeIDPrefix is the prefix for lingjun node IDs
 	LingjunNodeIDPrefix = "lingjun:"
 
 	// network types of CPFS mount targets

pkg/bmcpfs/controllerserver.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 
@@ -43,6 +44,7 @@ type controllerServer struct {
 	common.GenericControllerServer
 	vscManager     *internal.PrimaryVscManagerWithCache
 	attachDetacher internal.CPFSAttachDetacher
+	filsetManager  internal.CPFSFileSetManager
 	nasClient      *nasclient.Client
 	skipDetach     bool
 }
@@ -62,18 +64,85 @@ func newControllerServer(region string) (*controllerServer, error) {
 	if err != nil {
 		return nil, err
 	}
+
 	return &controllerServer{
 		vscManager:     internal.NewPrimaryVscManagerWithCache(efloClient),
 		attachDetacher: internal.NewCPFSAttachDetacher(nasClient),
+		filsetManager:  internal.NewCPFSFileSetManager(nasClient),
 		nasClient:      nasClient,
 		skipDetach:     skipDetach,
 	}, nil
 }
 
+// CreateVolume ...
+func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	logger := klog.FromContext(ctx)
+	logger.V(2).Info("starting")
+
+	// Validate parameters
+	if err := validateFileSetParameters(req); err != nil {
+		klog.Errorf("CreateVolume: error parameters from input: %v, with error: %v", req.Name, err)
+		return nil, status.Errorf(codes.InvalidArgument, "Invalid parameters from input: %v, with error: %v", req.Name, err)
+	}
+
+	// Extract parameters
+	params := req.GetParameters()
+	bmcpfsID := params["bmcpfsId"]
+	fullPath := req.Name
+	if rootPath, ok := params["path"]; ok && rootPath != "" {
+		fullPath = filepath.Join(rootPath, req.Name)
+	}
+	volSizeBytes := int64(req.GetCapacityRange().GetRequiredBytes())
+
+	// Create fileset
+	fileSetID, err := cs.filsetManager.CreateFileSet(ctx, bmcpfsID, req.Name, fullPath, 1000000, volSizeBytes, false)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	// Prepare volume context
+	volumeContext := req.GetParameters()
+	if volumeContext == nil {
+		volumeContext = make(map[string]string)
+	}
+	volumeContext = updateVolumeContext(volumeContext)
+
+	klog.Infof("CreateVolume: Successfully created FileSet %s: id[%s], filesystem[%s], path[%s]", req.GetName(), fileSetID, bmcpfsID, fullPath)
+
+	tmpVol := createVolumeResponse(fileSetID, bmcpfsID, volSizeBytes, volumeContext)
+
+	return &csi.CreateVolumeResponse{Volume: tmpVol}, nil
+}
+
+func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	logger := klog.FromContext(ctx)
+	logger.V(2).Info("starting")
+
+	// Parse volume ID to extract filesystem ID and fileset ID
+	fsID, fileSetID, err := parseVolumeID(req.VolumeId)
+	if err != nil {
+		klog.Errorf("DeleteVolume: failed to parse volume ID %s: %v", req.VolumeId, err)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	klog.Infof("DeleteVolume: deleting fileset %s from filesystem %s", fileSetID, fsID)
+
+	// Delete the fileset
+	err = cs.filsetManager.DeleteFileSet(ctx, fsID, fileSetID)
+	if err != nil {
+		klog.Errorf("DeleteVolume: failed to delete fileset %s from filesystem %s: %v", fileSetID, fsID, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	klog.Infof("DeleteVolume: successfully deleted fileset %s from filesystem %s", fileSetID, fsID)
+	return &csi.DeleteVolumeResponse{}, nil
+}
+
 func (cs *controllerServer) ControllerGetCapabilities(ctx context.Context, req *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
 	return &csi.ControllerGetCapabilitiesResponse{
 		Capabilities: common.ControllerRPCCapabilities(
 			csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
+			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		),
 	}, nil
 }
@@ -94,7 +163,7 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 	}
 
 	cpfsID, _ := parseVolumeHandle(req.VolumeId)
-	// Get VscMountTarget of filesystem
+
 	mt := req.VolumeContext[_vscMountTarget]
 	if mt == "" {
 		var err error
@@ -105,18 +174,18 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 	}
 
 	// Get Primary vsc of Lingjun node
-	lingjunInstanceId := strings.TrimPrefix(req.NodeId, LingjunNodeIDPrefix)
+	lingjunInstanceID := strings.TrimPrefix(req.NodeId, LingjunNodeIDPrefix)
 	if LingjunNodeIDPrefix == "" {
 		return nil, status.Error(codes.InvalidArgument, "invalid node id")
 	}
-	vscId, err := cs.vscManager.EnsurePrimaryVsc(ctx, lingjunInstanceId, false)
+	vscID, err := cs.vscManager.EnsurePrimaryVsc(ctx, lingjunInstanceID, false)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
-	klog.Info("Use VSC MountTarget for lingjun node", "nodeId", req.NodeId, "vscId", vscId)
+	klog.Info("Use VSC MountTarget for lingjun node", "nodeId", req.NodeId, "vscId", vscID)
 
 	// Attach CPFS to VSC
-	err = cs.attachDetacher.Attach(ctx, cpfsID, 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] == "" {
@@ -135,35 +204,34 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 
 	// TODO: if the cached vscid is already deleted, try to recreate a new primary vsc for lingjun node
 
-	klog.InfoS("ControllerPublishVolume: attached cpfs to vsc", "vscMountTarget", mt, "vscId", vscId, "node", req.NodeId)
+	klog.InfoS("ControllerPublishVolume: attached cpfs to vsc", "vscMountTarget", mt, "vscId", vscID, "node", req.NodeId)
 	return &csi.ControllerPublishVolumeResponse{
 		PublishContext: map[string]string{
 			_networkType:    networkTypeVSC,
-			_vscId:          vscId,
+			_vscID:          vscID,
 			_vscMountTarget: mt,
 		},
 	}, nil
 }
 
-func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (
-	*csi.ControllerUnpublishVolumeResponse, error,
-) {
+func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
 	if !strings.HasPrefix(req.NodeId, LingjunNodeIDPrefix) || cs.skipDetach {
 		return &csi.ControllerUnpublishVolumeResponse{}, nil
 	}
 	// Create Primary vsc for Lingjun node
-	lingjunInstanceId := strings.TrimPrefix(req.NodeId, LingjunNodeIDPrefix)
+	lingjunInstanceID := strings.TrimPrefix(req.NodeId, LingjunNodeIDPrefix)
 	if LingjunNodeIDPrefix == "" {
 		return nil, status.Error(codes.InvalidArgument, "invalid node id")
 	}
-	vsc, err := cs.vscManager.GetPrimaryVscOf(lingjunInstanceId)
+	vsc, err := cs.vscManager.GetPrimaryVscOf(lingjunInstanceID)
 	if err != nil {
-		return nil, status.Error(codes.Internal, err.Error())
+		return nil, status.Errorf(codes.Internal, "get vsc error: %v", err)
 	}
 	if vsc == nil {
 		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 {
@@ -173,6 +241,7 @@ func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *
 	return &csi.ControllerUnpublishVolumeResponse{}, nil
 }
 
+// KubernetesAlicloudIdentity is the user agent string for Eflo client
 var KubernetesAlicloudIdentity = fmt.Sprintf("Kubernetes.Alicloud/CsiProvision.Bmcpfs-%s", version.VERSION)
 
 const efloConnTimeout = 10
@@ -227,9 +296,9 @@ func parseVolumeHandle(volumeHandle string) (string, string) {
 	return parts[0], ""
 }
 
-func getMountTarget(client *nasclient.Client, fsId, networkType string) (string, error) {
+func getMountTarget(client *nasclient.Client, fsID, networkType string) (string, error) {
 	resp, err := client.DescribeFileSystems(&nasclient.DescribeFileSystemsRequest{
-		FileSystemId: &fsId,
+		FileSystemId: &fsID,
 	})
 	if err != nil {
 		return "", fmt.Errorf("nas:DescribeFileSystems failed: %w", err)
@@ -252,7 +321,7 @@ func getMountTarget(client *nasclient.Client, fsId, networkType string) (string,
 		if t == networkType {
 			mountTarget := tea.StringValue(mt.MountTargetDomain)
 			status := tea.StringValue(mt.Status)
-			klog.V(2).InfoS("Found cpfs mount target", "filesystem", fsId, "networkType", networkType, "mountTarget", mountTarget, "status", status)
+			klog.V(2).InfoS("Found cpfs mount target", "filesystem", fsID, "networkType", networkType, "mountTarget", mountTarget, "status", status)
 			if status == "Active" {
 				return mountTarget, nil
 			}

pkg/bmcpfs/controllerserver_test.go

@@ -19,6 +19,12 @@ package bmcpfs
 import (
 	"testing"
 
+	"context"
+	"errors"
+
+	nasclient "github.com/alibabacloud-go/nas-20170626/v4/client"
+	"github.com/golang/mock/gomock"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/bmcpfs/internal"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -27,34 +33,102 @@ func TestParseVolumeHandle(t *testing.T) {
 	tests := []struct {
 		name           string
 		volumeHandle   string
-		expectedFsId   string
-		expectedFsetId string
+		expectedFsID   string
+		expectedFsetID string
 	}{
 		{
 			name:           "single part",
 			volumeHandle:   "fs-12345",
-			expectedFsId:   "fs-12345",
-			expectedFsetId: "",
+			expectedFsID:   "fs-12345",
+			expectedFsetID: "",
 		},
 		{
 			name:           "two parts",
 			volumeHandle:   "fs-12345+fset-67890",
-			expectedFsId:   "fs-12345",
-			expectedFsetId: "fset-67890",
+			expectedFsID:   "fs-12345",
+			expectedFsetID: "fset-67890",
 		},
 		{
 			name:           "multiple delimiters",
 			volumeHandle:   "fs-12345+fset-67890+extra",
-			expectedFsId:   "fs-12345",
-			expectedFsetId: "",
+			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)
+			fsID, fsetID := parseVolumeHandle(tt.volumeHandle)
+			assert.Equal(t, tt.expectedFsID, fsID)
+			assert.Equal(t, tt.expectedFsetID, fsetID)
 		})
 	}
 }
+
+func TestControllerServer_DeleteVolume(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mockNas := internal.NewGoMockNasClient(ctrl)
+
+	mockNas.EXPECT().
+		DeleteFileset(gomock.Any()).
+		DoAndReturn(func(req *nasclient.DeleteFilesetRequest) (*nasclient.DeleteFilesetResponse, error) {
+			if *req.FileSystemId == "" {
+				return nil, errors.New("file system ID is required")
+			}
+			if *req.FsetId == "" {
+				return nil, errors.New("file set ID is required")
+			}
+			if *req.FileSystemId == "normal_delete" {
+				return &nasclient.DeleteFilesetResponse{
+					Body: &nasclient.DeleteFilesetResponseBody{},
+				}, nil
+			}
+			if *req.FileSystemId == "delete_failed" {
+				return nil, errors.New("delete failed")
+			}
+			return nil, nil
+		})
+
+	fsm := internal.NewCPFSFileSetManager(mockNas)
+	ctx := context.Background()
+
+	tests := []struct {
+		name        string
+		fsID        string
+		filesetID   string
+		expectError string
+	}{
+		{
+			name:        "filesystem empty",
+			fsID:        "",
+			filesetID:   "",
+			expectError: "file system ID is required",
+		},
+		{
+			name:        "fileset empty",
+			fsID:        "xxx",
+			filesetID:   "",
+			expectError: "file set ID is required",
+		},
+		{
+			name:        "normal delete",
+			fsID:        "normal_delete",
+			filesetID:   "normal_delete",
+			expectError: "",
+		},
+		{
+			name:        "delete_failed",
+			fsID:        "delete_failed",
+			filesetID:   "delete_failed",
+			expectError: "delete failed",
+		},
+	}
+
+	for _, tt := range tests {
+		err := fsm.DeleteFileSet(ctx, tt.fsID, tt.filesetID)
+		assert.Equal(t, tt.expectError, err.Error())
+	}
+
+}