Support fileset of bmcpfs

Author: mowangdk

_typos.toml

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

pkg/bmcpfs/bmcpfs.go

@@ -30,25 +30,28 @@ 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
 	networkTypeVPC = "vpc"
 	networkTypeVSC = "vsc"
 )
 
+// Driver represents the BMCPFS CSI driver
 type Driver struct {
 	endpoint string
 	servers  common.Servers
 }
 
+// NewDriver creates a new driver instance
 func NewDriver(meta *metadata.Metadata, endpoint string, serviceType utils.ServiceType) *Driver {
 	var driver Driver
 	driver.endpoint = endpoint
@@ -72,6 +75,7 @@ func NewDriver(meta *metadata.Metadata, endpoint string, serviceType utils.Servi
 	return &driver
 }
 
+// Run starts the driver
 func (d *Driver) Run() {
 	klog.InfoS("Starting driver", "driver", driverType, "endpoint", d.endpoint)
 	common.RunCSIServer(driverType, d.endpoint, d.servers)

pkg/bmcpfs/controllerserver.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 
@@ -42,6 +43,7 @@ type controllerServer struct {
 	common.GenericControllerServer
 	vscManager     *internal.PrimaryVscManagerWithCache
 	attachDetacher internal.CPFSAttachDetacher
+	filsetManager  internal.CPFSFileSetManager
 	nasClient      *nasclient.Client
 }
 
@@ -58,14 +60,98 @@ func newControllerServer(region string) (*controllerServer, error) {
 	return &controllerServer{
 		vscManager:     internal.NewPrimaryVscManagerWithCache(efloClient),
 		attachDetacher: internal.NewCPFSAttachDetacher(nasClient),
+		filsetManager:  internal.NewCPFSFileSetManager(nasClient),
 		nasClient:      nasClient,
 	}, nil
 }
 
+// CreateVolume ...
+func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	logger := klog.FromContext(ctx)
+	logger.V(2).Info("starting")
+
+	// Handle static volume creation (existing fileset)
+	csiVolume, err := staticFileSetCreate(req, cs.filsetManager)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "create static volume failed: %v", err)
+	}
+	if csiVolume != nil {
+		klog.Infof("CreateVolume: static volume create successful, pvName: %s, VolumeId: %s, volumeContext: %v", req.Name, csiVolume.VolumeId, csiVolume.VolumeContext)
+		return &csi.CreateVolumeResponse{Volume: csiVolume}, nil
+	}
+
+	// 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)
+
+	// Check if fileset exists before attempting to delete
+	_, err = cs.filsetManager.DescribeFileset(ctx, fsID, fileSetID)
+	if err != nil {
+		// If fileset doesn't exist, consider it already deleted
+		klog.V(2).Infof("DeleteVolume: fileset %s not found in filesystem %s, considering it already deleted", fileSetID, fsID)
+		return &csi.DeleteVolumeResponse{}, nil
+	}
+
+	// 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
 }
@@ -96,18 +182,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, req.VolumeId, vscId)
+	err = cs.attachDetacher.Attach(ctx, req.VolumeId, vscID)
 	if err != nil {
 		if autoSwitch, _ := strconv.ParseBool(req.VolumeContext[_mpAutoSwitch]); autoSwitch && internal.IsAttachNotSupportedError(err) {
 			if req.VolumeContext[_vpcMountTarget] == "" {
@@ -125,11 +211,11 @@ 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
@@ -143,11 +229,11 @@ func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *
 	}
 
 	// 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())
 	}
@@ -163,6 +249,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
@@ -209,9 +296,9 @@ func newEfloClient(region string) (*efloclient.Client, error) {
 	return efloclient.NewClient(config)
 }
 
-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)
@@ -234,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
 			}