feat: use file metadata for vhd disk lock

kubernetes-sigs · Mar 9, 2020 · 2fc4d7d · 2fc4d7d
1 parent b7567a0
commit 2fc4d7d
Show file tree

Hide file tree

Showing 2 changed files with 89 additions and 4 deletions.
diff --git a/pkg/azurefile/azurefile.go b/pkg/azurefile/azurefile.go
@@ -70,6 +70,7 @@ const (
 	fsTypeField   = "fstype"
 	proxyMount    = "proxy-mount"
 	cifs          = "cifs"
+	metaDataNode  = "node"
 )
 
 // Driver implements all interfaces of CSI drivers
@@ -112,6 +113,7 @@ func (d *Driver) Run(endpoint string) {
 	d.AddControllerServiceCapabilities(
 		[]csi.ControllerServiceCapability_RPC_Type{
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+			csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 			//csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 			csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,

diff --git a/pkg/azurefile/controllerserver.go b/pkg/azurefile/controllerserver.go
@@ -214,15 +214,98 @@ func (d *Driver) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (
 }
 
 // ControllerPublishVolume make a volume available on some required node
-// N/A for azure file
 func (d *Driver) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	klog.V(2).Infof("ControllerPublishVolume: called with args %+v", *req)
+	volumeID := req.GetVolumeId()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID not provided")
+	}
+
+	nodeID := req.GetNodeId()
+	if len(nodeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Node ID not provided")
+	}
+
+	resourceGroupName, accountName, fileShareName, diskName, err := getFileShareInfo(volumeID)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("getFileShareInfo(%s) failed with error: %v", volumeID, err))
+	}
+	if diskName == "" {
+		klog.V(2).Infof("skip ControllerPublishVolume process since disk name is empty, volumeid: %s", volumeID)
+		return &csi.ControllerPublishVolumeResponse{}, nil
+	}
+
+	accountKey, err := d.cloud.GetStorageAccesskey(accountName, resourceGroupName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("could not find key for storage account(%s) under resource group(%s), err %v", accountName, resourceGroupName, err))
+	}
+
+	storageEndpointSuffix := d.cloud.Environment.StorageEndpointSuffix
+	fileURL, err := getFileURL(accountName, accountKey, storageEndpointSuffix, fileShareName, diskName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("getFileURL(%s,%s,%s,%s) returned with error: %v", accountName, storageEndpointSuffix, fileShareName, diskName, err))
+	}
+	if fileURL == nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("getFileURL(%s,%s,%s,%s) return empty fileURL", accountName, storageEndpointSuffix, fileShareName, diskName))
+	}
+
+	properties, err := fileURL.GetProperties(ctx)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("GetProperties for volumeid(%s) on node(%s) returned with error: %v", volumeID, nodeID, err))
+	}
+
+	if v, ok := properties.NewMetadata()[metaDataNode]; ok {
+		if v != "" {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("volumeid(%s) cannot be attached to node(%s) since it's already attached to node(%s)", volumeID, nodeID, v))
+		}
+	}
+
+	if _, err = fileURL.SetMetadata(ctx, azfile.Metadata{metaDataNode: nodeID}); err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("SetMetadata for volumeid(%s) on node(%s) returned with error: %v", volumeID, nodeID, err))
+	}
+	return &csi.ControllerPublishVolumeResponse{}, nil
 }
 
 // ControllerUnpublishVolume make the volume unavailable on a specified node
-// N/A for azure file
 func (d *Driver) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	klog.V(2).Infof("ControllerUnpublishVolume: called with args %+v", *req)
+	volumeID := req.GetVolumeId()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID not provided")
+	}
+
+	nodeID := req.GetNodeId()
+	if len(nodeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Node ID not provided")
+	}
+
+	resourceGroupName, accountName, fileShareName, diskName, err := getFileShareInfo(volumeID)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("getFileShareInfo(%s) failed with error: %v", volumeID, err))
+	}
+	if diskName == "" {
+		klog.V(2).Infof("skip ControllerUnpublishVolume process since disk name is empty, volumeid: %s", volumeID)
+		return &csi.ControllerUnpublishVolumeResponse{}, nil
+	}
+
+	accountKey, err := d.cloud.GetStorageAccesskey(accountName, resourceGroupName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("could not find key for storage account(%s) under resource group(%s), err %v", accountName, resourceGroupName, err))
+	}
+
+	storageEndpointSuffix := d.cloud.Environment.StorageEndpointSuffix
+	fileURL, err := getFileURL(accountName, accountKey, storageEndpointSuffix, fileShareName, diskName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("getFileURL(%s,%s,%s,%s) returned with error: %v", accountName, storageEndpointSuffix, fileShareName, diskName, err))
+	}
+	if fileURL == nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("getFileURL(%s,%s,%s,%s) return empty fileURL", accountName, storageEndpointSuffix, fileShareName, diskName))
+	}
+
+	if _, err = fileURL.SetMetadata(ctx, azfile.Metadata{metaDataNode: ""}); err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("SetMetadata for volumeid(%s) on node(%s) returned with error: %v", volumeID, nodeID, err))
+	}
+	return &csi.ControllerUnpublishVolumeResponse{}, nil
 }
 
 // CreateSnapshot create a snapshot (todo)