Repository: kubernetes-sigs/cloud-provider-azure
Pull request #453 (2020-12-28 12:16:13)

use batch operation for azure disk attach/detach
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/release.md#issue-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
6. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
/kind feature

**What this PR does / why we need it**:
use batch operation for azure disk attach/detach
This PR use batch operation for disk attach/detach, that would improve disk attach/detach operation speed quite a lot, e.g. if we attach 20 disks to one vmss instance at one time, with this PR, only two disk attach operations would be triggered, that's 10x faster than original.

 - For attach disk operation, there is a queue called `attachDiskMap `
 - For detach disk operation, there is a queue called `detachDiskMap `


```
	// disk queue that is waiting for attach or detach on specific node
	// <nodeName, map<diskURI, *AttachDiskOptions/DetachDiskOptions>>
	attachDiskMap sync.Map
	detachDiskMap sync.Map
```

**Which issue(s) this PR fixes**:
<!-- 
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:
move  pkg/mockvmsets to provider folder since there is a cycle import error if not move:
```
# sigs.k8s.io/cloud-provider-azure/pkg/provider
package sigs.k8s.io/cloud-provider-azure/pkg/provider (test)
        imports sigs.k8s.io/cloud-provider-azure/pkg/mockvmsets
        imports sigs.k8s.io/cloud-provider-azure/pkg/provider: import cycle not allowed in test
FAIL    sigs.k8s.io/cloud-provider-azure/pkg/provider [setup failed]
```

**Release note**:
```
use batch operation for azure disk attach/detach
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/453#" title="Author self-approved">andyzhangx</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fcloud-provider-azure).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/cloud-provider-azure/blob/master/OWNERS)~~ [andyzhangx]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
andyzhangx: /hold
andyzhangx: hold for review and testing on CSI driver first, testing result on in-tree driver is quite positive, will port this code change to CSI driver also.
k8s-ci-robot: @andyzhangx: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm/1343531202862125056) | `/test pull-cloud-provider-azure-e2e-ccm`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=453). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Aandyzhangx). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @andyzhangx: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm/1343531202862125056) | `/test pull-cloud-provider-azure-e2e-ccm`
pull-kubernetes-e2e-azure-disk-vmss-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-kubernetes-e2e-azure-disk-vmss-ccm/1343531202891485184) | `/test pull-kubernetes-e2e-azure-disk-vmss-ccm`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=453). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Aandyzhangx). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @andyzhangx: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm/1343531202862125056) | `/test pull-cloud-provider-azure-e2e-ccm`
pull-kubernetes-e2e-azure-disk-vmss-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-kubernetes-e2e-azure-disk-vmss-ccm/1343531202891485184) | `/test pull-kubernetes-e2e-azure-disk-vmss-ccm`
pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb/1343531202836959232) | `/test pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=453). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Aandyzhangx). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
andyzhangx: /retest
k8s-ci-robot: @andyzhangx: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-e2e-azure-disk-vmss-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-kubernetes-e2e-azure-disk-vmss-ccm/1343531202891485184) | `/test pull-kubernetes-e2e-azure-disk-vmss-ccm`
pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb/1343531202836959232) | `/test pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb`
pull-cloud-provider-azure-e2e-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm/1343733997062066176) | `/test pull-cloud-provider-azure-e2e-ccm`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=453). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Aandyzhangx). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
nilo19: for the basic lb test failure: https://github.com/kubernetes-sigs/cloud-provider-azure/pull/455
nilo19: need to investigate why the tests cluster cannot be built in pull-cloud-provider-azure-e2e-ccm
k8s-ci-robot: @andyzhangx: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-kubernetes-e2e-azure-disk-vmss-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-kubernetes-e2e-azure-disk-vmss-ccm/1343531202891485184) | `/test pull-kubernetes-e2e-azure-disk-vmss-ccm`
pull-cloud-provider-azure-e2e-ccm | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm/1343733997062066176) | `/test pull-cloud-provider-azure-e2e-ccm`
pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb | c2d20d5bd4c69ed38605f98e0592879773260c39 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/453/pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb/1343733997041094656) | `/test pull-cloud-provider-azure-e2e-ccm-vmss-basic-lb`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=453). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Aandyzhangx). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/pkg/mockvmsets/BUILD.bazel b/pkg/mockvmsets/BUILD.bazel
deleted file mode 100644
index 85db25a60..000000000
--- a/pkg/mockvmsets/BUILD.bazel
+++ /dev/null
@@ -1,20 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "azure_mock_vmsets.go",
-        "doc.go",
-    ],
-    importpath = "sigs.k8s.io/cloud-provider-azure/pkg/mockvmsets",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/cache:go_default_library",
-        "//vendor/github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-06-30/compute:go_default_library",
-        "//vendor/github.com/Azure/azure-sdk-for-go/services/network/mgmt/2019-06-01/network:go_default_library",
-        "//vendor/github.com/golang/mock/gomock:go_default_library",
-        "//vendor/k8s.io/api/core/v1:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
-        "//vendor/k8s.io/cloud-provider:go_default_library",
-    ],
-)
diff --git a/pkg/mockvmsets/doc.go b/pkg/mockvmsets/doc.go
deleted file mode 100644
index 2caad5d48..000000000
--- a/pkg/mockvmsets/doc.go
+++ /dev/null
@@ -1,20 +0,0 @@
-// +build !providerless
-
-/*
-Copyright 2020 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 mockvmsets implements the mock client for VMSet.
-package mockvmsets // import "sigs.k8s.io/cloud-provider-azure/pkg/mockvmsets"
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index e9d6c0702..774c33c88 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -726,7 +726,7 @@ func initDiskControllers(az *Cloud) error {
 		resourceGroup:         az.ResourceGroup,
 		subscriptionID:        az.SubscriptionID,
 		cloud:                 az,
-		vmLockMap:             newLockMap(),
+		lockMap:               newLockMap(),
 	}
 
 	az.BlobDiskController = &BlobDiskController{common: common}
diff --git a/pkg/provider/azure_controller_common.go b/pkg/provider/azure_controller_common.go
index 601a868f0..28abb6a55 100644
--- a/pkg/provider/azure_controller_common.go
+++ b/pkg/provider/azure_controller_common.go
@@ -47,14 +47,16 @@ const (
 	// https://docs.microsoft.com/en-us/azure/virtual-machines/premium-storage-performance#disk-caching
 	diskCachingLimit = 4096 // GiB
 
-	maxLUN               = 64 // max number of LUNs per VM
-	errLeaseIDMissing    = "LeaseIdMissing"
-	errContainerNotFound = "ContainerNotFound"
-	errStatusCode400     = "statuscode=400"
-	errInvalidParameter  = `code="invalidparameter"`
-	errTargetInstanceIds = `target="instanceids"`
-	sourceSnapshot       = "snapshot"
-	sourceVolume         = "volume"
+	maxLUN                 = 64 // max number of LUNs per VM
+	errLeaseIDMissing      = "LeaseIdMissing"
+	errContainerNotFound   = "ContainerNotFound"
+	errStatusCode400       = "statuscode=400"
+	errInvalidParameter    = `code="invalidparameter"`
+	errTargetInstanceIds   = `target="instanceids"`
+	sourceSnapshot         = "snapshot"
+	sourceVolume           = "volume"
+	attachDiskMapKeySuffix = "attachdiskmap"
+	detachDiskMapKeySuffix = "detachdiskmap"
 
 	// WriteAcceleratorEnabled support for Azure Write Accelerator on Azure Disks
 	// https://docs.microsoft.com/azure/virtual-machines/windows/how-to-enable-write-accelerator
@@ -84,11 +86,23 @@ type controllerCommon struct {
 	location              string
 	storageEndpointSuffix string
 	resourceGroup         string
-	// store disk URI when disk is in attaching or detaching process
-	diskAttachDetachMap sync.Map
-	// vm disk map used to lock per vm update calls
-	vmLockMap *lockMap
-	cloud     *Cloud
+	diskStateMap          sync.Map // <diskURI, attaching/detaching state>
+	lockMap               *lockMap
+	cloud                 *Cloud
+	// disk queue that is waiting for attach or detach on specific node
+	// <nodeName, map<diskURI, *AttachDiskOptions/DetachDiskOptions>>
+	attachDiskMap sync.Map
+	detachDiskMap sync.Map
+}
+
+// AttachDiskOptions attach disk options
+type AttachDiskOptions struct {
+	lun                     int32
+	isManagedDisk           bool
+	diskName                string
+	cachingMode             compute.CachingTypes
+	diskEncryptionSetID     string
+	writeAcceleratorEnabled bool
 }
 
 // getNodeVMSet gets the VMSet interface based on config.VMType and the real virtual machine type.
@@ -179,28 +193,85 @@ func (c *controllerCommon) AttachDisk(isManagedDisk bool, diskName, diskURI stri
 		}
 	}
 
-	instanceid, err := c.cloud.InstanceID(context.TODO(), nodeName)
-	if err != nil {
-		klog.Warningf("failed to get azure instance id (%v) for node %s", err, nodeName)
-		return -1, fmt.Errorf("failed to get azure instance id for node %q (%v)", nodeName, err)
+	options := AttachDiskOptions{
+		lun:                     -1,
+		isManagedDisk:           isManagedDisk,
+		diskName:                diskName,
+		cachingMode:             cachingMode,
+		diskEncryptionSetID:     diskEncryptionSetID,
+		writeAcceleratorEnabled: writeAcceleratorEnabled,
+	}
+	node := strings.ToLower(string(nodeName))
+	disk := strings.ToLower(string(diskURI))
+	if err := c.insertAttachDiskRequest(disk, node, &options); err != nil {
+		return -1, err
 	}
 
-	c.vmLockMap.LockEntry(strings.ToLower(string(nodeName)))
-	defer c.vmLockMap.UnlockEntry(strings.ToLower(string(nodeName)))
+	c.lockMap.LockEntry(node)
+	defer c.lockMap.UnlockEntry(node)
+	diskMap, err := c.cleanAttachDiskRequests(node)
+	if err != nil {
+		return -1, err
+	}
 
-	lun, err := c.GetNextDiskLun(nodeName)
+	lun, err := c.SetDiskLun(nodeName, disk, diskMap)
 	if err != nil {
-		klog.Warningf("no LUN available for instance %q (%v)", nodeName, err)
-		return -1, fmt.Errorf("all LUNs are used, cannot attach volume (%s, %s) to instance %q (%v)", diskName, diskURI, instanceid, err)
+		return -1, err
+	}
+
+	klog.V(2).Infof("Trying to attach volume %q lun %d to node %q, diskMap: %s", diskURI, lun, nodeName, diskMap)
+	if len(diskMap) == 0 {
+		return lun, nil
+	}
+	c.diskStateMap.Store(disk, "attaching")
+	defer c.diskStateMap.Delete(disk)
+	return lun, vmset.AttachDisk(nodeName, diskMap)
+}
+
+func (c *controllerCommon) insertAttachDiskRequest(diskURI, nodeName string, options *AttachDiskOptions) error {
+	var diskMap map[string]*AttachDiskOptions
+	attachDiskMapKey := nodeName + attachDiskMapKeySuffix
+	c.lockMap.LockEntry(attachDiskMapKey)
+	defer c.lockMap.UnlockEntry(attachDiskMapKey)
+	v, ok := c.attachDiskMap.Load(nodeName)
+	if ok {
+		if diskMap, ok = v.(map[string]*AttachDiskOptions); !ok {
+			return fmt.Errorf("convert attachDiskMap failure on node(%s)", nodeName)
+		}
+	} else {
+		diskMap = make(map[string]*AttachDiskOptions)
+		c.attachDiskMap.Store(nodeName, diskMap)
+	}
+	// insert attach disk request to queue
+	_, ok = diskMap[diskURI]
+	if ok {
+		klog.V(2).Infof("azureDisk - duplicated attach disk(%s) request on node(%s)", diskURI, nodeName)
+	} else {
+		diskMap[diskURI] = options
 	}
+	return nil
+}
 
-	klog.V(2).Infof("Trying to attach volume %q lun %d to node %q.", diskURI, lun, nodeName)
-	c.diskAttachDetachMap.Store(strings.ToLower(diskURI), "attaching")
-	defer c.diskAttachDetachMap.Delete(strings.ToLower(diskURI))
-	return lun, vmset.AttachDisk(isManagedDisk, diskName, diskURI, nodeName, lun, cachingMode, diskEncryptionSetID, writeAcceleratorEnabled)
+// clean up attach disk requests
+// return original attach disk requests
+func (c *controllerCommon) cleanAttachDiskRequests(nodeName string) (map[string]*AttachDiskOptions, error) {
+	var diskMap map[string]*AttachDiskOptions
+
+	attachDiskMapKey := nodeName + attachDiskMapKeySuffix
+	c.lockMap.LockEntry(attachDiskMapKey)
+	defer c.lockMap.UnlockEntry(attachDiskMapKey)
+	v, ok := c.attachDiskMap.Load(nodeName)
+	if !ok {
+		return diskMap, nil
+	}
+	if diskMap, ok = v.(map[string]*AttachDiskOptions); !ok {
+		return diskMap, fmt.Errorf("convert attachDiskMap failure on node(%s)", nodeName)
+	}
+	c.attachDiskMap.Store(nodeName, make(map[string]*AttachDiskOptions))
+	return diskMap, nil
 }
 
-// DetachDisk detaches a disk from host. The vhd can be identified by diskName or diskURI.
+// DetachDisk detaches a disk from VM
 func (c *controllerCommon) DetachDisk(diskName, diskURI string, nodeName types.NodeName) error {
 	_, err := c.cloud.InstanceID(context.TODO(), nodeName)
 	if err != nil {
@@ -219,14 +290,25 @@ func (c *controllerCommon) DetachDisk(diskName, diskURI string, nodeName types.N
 		return err
 	}
 
-	klog.V(2).Infof("detach %v from node %q", diskURI, nodeName)
+	node := strings.ToLower(string(nodeName))
+	disk := strings.ToLower(string(diskURI))
+	if err := c.insertDetachDiskRequest(diskName, disk, node); err != nil {
+		return err
+	}
 
-	// make the lock here as small as possible
-	c.vmLockMap.LockEntry(strings.ToLower(string(nodeName)))
-	c.diskAttachDetachMap.Store(strings.ToLower(diskURI), "detaching")
-	err = vmset.DetachDisk(diskName, diskURI, nodeName)
-	c.diskAttachDetachMap.Delete(strings.ToLower(diskURI))
-	c.vmLockMap.UnlockEntry(strings.ToLower(string(nodeName)))
+	c.lockMap.LockEntry(node)
+	diskMap, err := c.cleanDetachDiskRequests(node)
+	if err != nil {
+		return err
+	}
+
+	klog.V(2).Infof("Trying to detach volume %q from node %q, diskMap: %s", diskURI, nodeName, diskMap)
+	if len(diskMap) > 0 {
+		c.diskStateMap.Store(disk, "detaching")
+		err = vmset.DetachDisk(nodeName, diskMap)
+		c.diskStateMap.Delete(disk)
+	}
+	c.lockMap.UnlockEntry(node)
 
 	if err != nil {
 		if isInstanceNotFoundError(err) {
@@ -238,11 +320,11 @@ func (c *controllerCommon) DetachDisk(diskName, diskURI string, nodeName types.N
 		if retry.IsErrorRetriable(err) && c.cloud.CloudProviderBackoff {
 			klog.Warningf("azureDisk - update backing off: detach disk(%s, %s), err: %v", diskName, diskURI, err)
 			retryErr := kwait.ExponentialBackoff(c.cloud.RequestBackoff(), func() (bool, error) {
-				c.vmLockMap.LockEntry(strings.ToLower(string(nodeName)))
-				c.diskAttachDetachMap.Store(strings.ToLower(diskURI), "detaching")
-				err := vmset.DetachDisk(diskName, diskURI, nodeName)
-				c.diskAttachDetachMap.Delete(strings.ToLower(diskURI))
-				c.vmLockMap.UnlockEntry(strings.ToLower(string(nodeName)))
+				c.lockMap.LockEntry(node)
+				c.diskStateMap.Store(disk, "detaching")
+				err := vmset.DetachDisk(nodeName, diskMap)
+				c.diskStateMap.Delete(disk)
+				c.lockMap.UnlockEntry(node)
 
 				retriable := false
 				if err != nil && retry.IsErrorRetriable(err) {
@@ -265,6 +347,50 @@ func (c *controllerCommon) DetachDisk(diskName, diskURI string, nodeName types.N
 	return nil
 }
 
+func (c *controllerCommon) insertDetachDiskRequest(diskName, diskURI, nodeName string) error {
+	var diskMap map[string]string
+	detachDiskMapKey := nodeName + detachDiskMapKeySuffix
+	c.lockMap.LockEntry(detachDiskMapKey)
+	defer c.lockMap.UnlockEntry(detachDiskMapKey)
+	v, ok := c.detachDiskMap.Load(nodeName)
+	if ok {
+		if diskMap, ok = v.(map[string]string); !ok {
+			return fmt.Errorf("convert detachDiskMap failure on node(%s)", nodeName)
+		}
+	} else {
+		diskMap = make(map[string]string)
+		c.detachDiskMap.Store(nodeName, diskMap)
+	}
+	// insert detach disk request to queue
+	_, ok = diskMap[diskURI]
+	if ok {
+		klog.V(2).Infof("azureDisk - duplicated detach disk(%s) request on node(%s)", diskURI, nodeName)
+	} else {
+		diskMap[diskURI] = diskName
+	}
+	return nil
+}
+
+// clean up detach disk requests
+// return original detach disk requests
+func (c *controllerCommon) cleanDetachDiskRequests(nodeName string) (map[string]string, error) {
+	var diskMap map[string]string
+
+	detachDiskMapKey := nodeName + detachDiskMapKeySuffix
+	c.lockMap.LockEntry(detachDiskMapKey)
+	defer c.lockMap.UnlockEntry(detachDiskMapKey)
+	v, ok := c.detachDiskMap.Load(nodeName)
+	if !ok {
+		return diskMap, nil
+	}
+	if diskMap, ok = v.(map[string]string); !ok {
+		return diskMap, fmt.Errorf("convert detachDiskMap failure on node(%s)", nodeName)
+	}
+	// clean up original requests in disk map
+	c.detachDiskMap.Store(nodeName, make(map[string]string))
+	return diskMap, nil
+}
+
 // getNodeDataDisks invokes vmSet interfaces to get data disks for the node.
 func (c *controllerCommon) getNodeDataDisks(nodeName types.NodeName, crt azcache.AzureCacheReadType) ([]compute.DataDisk, error) {
 	vmset, err := c.getNodeVMSet(nodeName, crt)
@@ -301,26 +427,70 @@ func (c *controllerCommon) GetDiskLun(diskName, diskURI string, nodeName types.N
 	return -1, fmt.Errorf("cannot find Lun for disk %s", diskName)
 }
 
-// GetNextDiskLun searches all vhd attachment on the host and find unused lun. Return -1 if all luns are used.
-func (c *controllerCommon) GetNextDiskLun(nodeName types.NodeName) (int32, error) {
-	disks, err := c.getNodeDataDisks(nodeName, azcache.CacheReadTypeDefault)
+// SetDiskLun find unused luns and allocate lun for every disk in diskMap.
+// Return lun of diskURI, -1 if all luns are used.
+func (c *controllerCommon) SetDiskLun(nodeName types.NodeName, diskURI string, diskMap map[string]*AttachDiskOptions) (int32, error) {
+	disks, err := c.getNodeDataDisks(nodeName, azcache.CacheReadTypeForceRefresh)
 	if err != nil {
 		klog.Errorf("error of getting data disks for node %q: %v", nodeName, err)
 		return -1, err
 	}
 
+	lun := int32(-1)
+	_, isDiskInMap := diskMap[diskURI]
 	used := make([]bool, maxLUN)
 	for _, disk := range disks {
 		if disk.Lun != nil {
 			used[*disk.Lun] = true
+			if !isDiskInMap {
+				// find lun of diskURI since diskURI is not in diskMap
+				if disk.ManagedDisk != nil && strings.EqualFold(*disk.ManagedDisk.ID, diskURI) {
+					lun = *disk.Lun
+				}
+			}
 		}
 	}
+	if !isDiskInMap && lun < 0 {
+		return -1, fmt.Errorf("could not find disk(%s) in current disk list(len: %d) nor in diskMap(%v)", diskURI, len(disks), diskMap)
+	}
+	if len(diskMap) == 0 {
+		// attach disk request is empty, return directly
+		return lun, nil
+	}
+
+	// allocate lun for every disk in diskMap
+	var diskLuns []int32
+	count := 0
 	for k, v := range used {
 		if !v {
-			return int32(k), nil
+			diskLuns = append(diskLuns, int32(k))
+			count++
+			if count >= len(diskMap) {
+				break
+			}
 		}
 	}
-	return -1, fmt.Errorf("all luns are used")
+
+	if len(diskLuns) != len(diskMap) {
+		return -1, fmt.Errorf("could not find enough disk luns(current: %d) for diskMap(%v, len=%d), diskURI(%s)",
+			len(diskLuns), diskMap, len(diskMap), diskURI)
+	}
+
+	count = 0
+	for uri, opt := range diskMap {
+		if opt == nil {
+			return -1, fmt.Errorf("unexpected nil pointer in diskMap(%v), diskURI(%s)", diskMap, diskURI)
+		}
+		if strings.EqualFold(uri, diskURI) {
+			lun = diskLuns[count]
+		}
+		opt.lun = diskLuns[count]
+		count++
+	}
+	if lun < 0 {
+		return lun, fmt.Errorf("could not find lun of diskURI(%s), diskMap(%v)", diskURI, diskMap)
+	}
+	return lun, nil
 }
 
 // DisksAreAttached checks if a list of volumes are attached to the node with the specified NodeName.
diff --git a/pkg/provider/azure_controller_standard.go b/pkg/provider/azure_controller_standard.go
index dac667599..72a847c58 100644
--- a/pkg/provider/azure_controller_standard.go
+++ b/pkg/provider/azure_controller_standard.go
@@ -29,9 +29,8 @@ import (
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 )
 
-// AttachDisk attaches a vhd to vm
-// the vhd must exist, can be identified by diskName, diskURI, and lun.
-func (as *availabilitySet) AttachDisk(isManagedDisk bool, diskName, diskURI string, nodeName types.NodeName, lun int32, cachingMode compute.CachingTypes, diskEncryptionSetID string, writeAcceleratorEnabled bool) error {
+// AttachDisk attaches a disk to vm
+func (as *availabilitySet) AttachDisk(nodeName types.NodeName, diskMap map[string]*AttachDiskOptions) error {
 	vm, err := as.getVirtualMachine(nodeName, azcache.CacheReadTypeDefault)
 	if err != nil {
 		return err
@@ -46,40 +45,56 @@ func (as *availabilitySet) AttachDisk(isManagedDisk bool, diskName, diskURI stri
 	disks := make([]compute.DataDisk, len(*vm.StorageProfile.DataDisks))
 	copy(disks, *vm.StorageProfile.DataDisks)
 
-	if isManagedDisk {
-		managedDisk := &compute.ManagedDiskParameters{ID: &diskURI}
-		if diskEncryptionSetID == "" {
-			if vm.StorageProfile.OsDisk != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID != nil {
-				// set diskEncryptionSet as value of os disk by default
-				diskEncryptionSetID = *vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID
+	for k, v := range diskMap {
+		diskURI := k
+		opt := v
+		if opt.isManagedDisk {
+			attached := false
+			for _, disk := range *vm.StorageProfile.DataDisks {
+				if disk.ManagedDisk != nil && strings.EqualFold(*disk.ManagedDisk.ID, diskURI) {
+					attached = true
+					break
+				}
 			}
+			if attached {
+				klog.V(2).Infof("azureDisk - disk(%s) already attached to node(%s)", diskURI, nodeName)
+				continue
+			}
+
+			managedDisk := &compute.ManagedDiskParameters{ID: &diskURI}
+			if opt.diskEncryptionSetID == "" {
+				if vm.StorageProfile.OsDisk != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID != nil {
+					// set diskEncryptionSet as value of os disk by default
+					opt.diskEncryptionSetID = *vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID
+				}
+			}
+			if opt.diskEncryptionSetID != "" {
+				managedDisk.DiskEncryptionSet = &compute.DiskEncryptionSetParameters{ID: &opt.diskEncryptionSetID}
+			}
+			disks = append(disks,
+				compute.DataDisk{
+					Name:                    &opt.diskName,
+					Lun:                     &opt.lun,
+					Caching:                 opt.cachingMode,
+					CreateOption:            "attach",
+					ManagedDisk:             managedDisk,
+					WriteAcceleratorEnabled: to.BoolPtr(opt.writeAcceleratorEnabled),
+				})
+		} else {
+			disks = append(disks,
+				compute.DataDisk{
+					Name: &opt.diskName,
+					Vhd: &compute.VirtualHardDisk{
+						URI: &diskURI,
+					},
+					Lun:          &opt.lun,
+					Caching:      opt.cachingMode,
+					CreateOption: "attach",
+				})
 		}
-		if diskEncryptionSetID != "" {
-			managedDisk.DiskEncryptionSet = &compute.DiskEncryptionSetParameters{ID: &diskEncryptionSetID}
-		}
-		disks = append(disks,
-			compute.DataDisk{
-				Name:                    &diskName,
-				Lun:                     &lun,
-				Caching:                 cachingMode,
-				CreateOption:            "attach",
-				ManagedDisk:             managedDisk,
-				WriteAcceleratorEnabled: to.BoolPtr(writeAcceleratorEnabled),
-			})
-	} else {
-		disks = append(disks,
-			compute.DataDisk{
-				Name: &diskName,
-				Vhd: &compute.VirtualHardDisk{
-					URI: &diskURI,
-				},
-				Lun:          &lun,
-				Caching:      cachingMode,
-				CreateOption: "attach",
-			})
 	}
 
 	newVM := compute.VirtualMachineUpdate{
@@ -89,7 +104,7 @@ func (as *availabilitySet) AttachDisk(isManagedDisk bool, diskName, diskURI stri
 			},
 		},
 	}
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk(%s, %s) with DiskEncryptionSetID(%s)", nodeResourceGroup, vmName, diskName, diskURI, diskEncryptionSetID)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk list(%s)", nodeResourceGroup, vmName, diskMap)
 	ctx, cancel := getContextWithCancel()
 	defer cancel()
 
@@ -100,29 +115,28 @@ func (as *availabilitySet) AttachDisk(isManagedDisk bool, diskName, diskURI stri
 
 	rerr := as.VirtualMachinesClient.Update(ctx, nodeResourceGroup, vmName, newVM, "attach_disk")
 	if rerr != nil {
-		klog.Errorf("azureDisk - attach disk(%s, %s) on rg(%s) vm(%s) failed, err: %v", diskName, diskURI, nodeResourceGroup, vmName, rerr)
+		klog.Errorf("azureDisk - attach disk list(%s) on rg(%s) vm(%s) failed, err: %v", diskMap, nodeResourceGroup, vmName, rerr)
 		if rerr.HTTPStatusCode == http.StatusNotFound {
-			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%s, %s) on rg(%s) vm(%s)", diskName, diskURI, nodeResourceGroup, vmName)
+			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%v) on rg(%s) vm(%s)", diskMap, nodeResourceGroup, vmName)
 			disks := as.filterNonExistingDisks(ctx, *newVM.VirtualMachineProperties.StorageProfile.DataDisks)
 			newVM.VirtualMachineProperties.StorageProfile.DataDisks = &disks
 			rerr = as.VirtualMachinesClient.Update(ctx, nodeResourceGroup, vmName, newVM, "attach_disk")
 		}
 	}
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk(%s, %s) returned with %v", nodeResourceGroup, vmName, diskName, diskURI, rerr)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk list(%s) returned with %v", nodeResourceGroup, vmName, diskMap, rerr)
 	if rerr != nil {
 		return rerr.Error()
 	}
 	return nil
 }
 
-// DetachDisk detaches a disk from host
-// the vhd can be identified by diskName or diskURI
-func (as *availabilitySet) DetachDisk(diskName, diskURI string, nodeName types.NodeName) error {
+// DetachDisk detaches a disk from VM
+func (as *availabilitySet) DetachDisk(nodeName types.NodeName, diskMap map[string]string) error {
 	vm, err := as.getVirtualMachine(nodeName, azcache.CacheReadTypeDefault)
 	if err != nil {
 		// if host doesn't exist, no need to detach
-		klog.Warningf("azureDisk - cannot find node %s, skip detaching disk(%s, %s)", nodeName, diskName, diskURI)
+		klog.Warningf("azureDisk - cannot find node %s, skip detaching disk list(%s)", nodeName, diskMap)
 		return nil
 	}
 
@@ -137,24 +151,25 @@ func (as *availabilitySet) DetachDisk(diskName, diskURI string, nodeName types.N
 
 	bFoundDisk := false
 	for i, disk := range disks {
-		if disk.Lun != nil && (disk.Name != nil && diskName != "" && strings.EqualFold(*disk.Name, diskName)) ||
-			(disk.Vhd != nil && disk.Vhd.URI != nil && diskURI != "" && strings.EqualFold(*disk.Vhd.URI, diskURI)) ||
-			(disk.ManagedDisk != nil && diskURI != "" && strings.EqualFold(*disk.ManagedDisk.ID, diskURI)) {
-			// found the disk
-			klog.V(2).Infof("azureDisk - detach disk: name %q uri %q", diskName, diskURI)
-			if strings.EqualFold(as.cloud.Environment.Name, AzureStackCloudName) {
-				disks = append(disks[:i], disks[i+1:]...)
-			} else {
-				disks[i].ToBeDetached = to.BoolPtr(true)
+		for diskURI, diskName := range diskMap {
+			if disk.Lun != nil && (disk.Name != nil && diskName != "" && strings.EqualFold(*disk.Name, diskName)) ||
+				(disk.Vhd != nil && disk.Vhd.URI != nil && diskURI != "" && strings.EqualFold(*disk.Vhd.URI, diskURI)) ||
+				(disk.ManagedDisk != nil && diskURI != "" && strings.EqualFold(*disk.ManagedDisk.ID, diskURI)) {
+				// found the disk
+				klog.V(2).Infof("azureDisk - detach disk: name %q uri %q", diskName, diskURI)
+				if strings.EqualFold(as.cloud.Environment.Name, AzureStackCloudName) {
+					disks = append(disks[:i], disks[i+1:]...)
+				} else {
+					disks[i].ToBeDetached = to.BoolPtr(true)
+				}
+				bFoundDisk = true
 			}
-			bFoundDisk = true
-			break
 		}
 	}
 
 	if !bFoundDisk {
 		// only log here, next action is to update VM status with original meta data
-		klog.Errorf("detach azure disk: disk %s not found, diskURI: %s", diskName, diskURI)
+		klog.Errorf("detach azure disk on node(%s): disk list(%s) not found", nodeName, diskMap)
 	}
 
 	newVM := compute.VirtualMachineUpdate{
@@ -164,7 +179,7 @@ func (as *availabilitySet) DetachDisk(diskName, diskURI string, nodeName types.N
 			},
 		},
 	}
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk(%s, %s)", nodeResourceGroup, vmName, diskName, diskURI)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk list(%s)", nodeResourceGroup, vmName, nodeName, diskMap)
 	ctx, cancel := getContextWithCancel()
 	defer cancel()
 
@@ -175,16 +190,16 @@ func (as *availabilitySet) DetachDisk(diskName, diskURI string, nodeName types.N
 
 	rerr := as.VirtualMachinesClient.Update(ctx, nodeResourceGroup, vmName, newVM, "detach_disk")
 	if rerr != nil {
-		klog.Errorf("azureDisk - detach disk(%s, %s) on rg(%s) vm(%s) failed, err: %v", diskName, diskURI, nodeResourceGroup, vmName, rerr)
+		klog.Errorf("azureDisk - detach disk list(%s) on rg(%s) vm(%s) failed, err: %v", diskMap, nodeResourceGroup, vmName, rerr)
 		if rerr.HTTPStatusCode == http.StatusNotFound {
-			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%s, %s) on rg(%s) vm(%s)", diskName, diskURI, nodeResourceGroup, vmName)
+			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%v) on rg(%s) vm(%s)", diskMap, nodeResourceGroup, vmName)
 			disks := as.filterNonExistingDisks(ctx, *vm.StorageProfile.DataDisks)
 			newVM.VirtualMachineProperties.StorageProfile.DataDisks = &disks
 			rerr = as.VirtualMachinesClient.Update(ctx, nodeResourceGroup, vmName, newVM, "detach_disk")
 		}
 	}
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk(%s, %s) returned with %v", nodeResourceGroup, vmName, diskName, diskURI, rerr)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk list(%s) returned with %v", nodeResourceGroup, vmName, diskMap, rerr)
 	if rerr != nil {
 		return rerr.Error()
 	}
diff --git a/pkg/provider/azure_controller_vmss.go b/pkg/provider/azure_controller_vmss.go
index 0f7ffd25a..da6dcaf02 100644
--- a/pkg/provider/azure_controller_vmss.go
+++ b/pkg/provider/azure_controller_vmss.go
@@ -29,9 +29,8 @@ import (
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 )
 
-// AttachDisk attaches a vhd to vm
-// the vhd must exist, can be identified by diskName, diskURI, and lun.
-func (ss *scaleSet) AttachDisk(isManagedDisk bool, diskName, diskURI string, nodeName types.NodeName, lun int32, cachingMode compute.CachingTypes, diskEncryptionSetID string, writeAcceleratorEnabled bool) error {
+// AttachDisk attaches a disk to vm
+func (ss *scaleSet) AttachDisk(nodeName types.NodeName, diskMap map[string]*AttachDiskOptions) error {
 	vmName := mapNodeNameToVMName(nodeName)
 	ssName, instanceID, vm, err := ss.getVmssVM(vmName, azcache.CacheReadTypeDefault)
 	if err != nil {
@@ -48,41 +47,59 @@ func (ss *scaleSet) AttachDisk(isManagedDisk bool, diskName, diskURI string, nod
 		disks = make([]compute.DataDisk, len(*vm.StorageProfile.DataDisks))
 		copy(disks, *vm.StorageProfile.DataDisks)
 	}
-	if isManagedDisk {
-		managedDisk := &compute.ManagedDiskParameters{ID: &diskURI}
-		if diskEncryptionSetID == "" {
-			if vm.StorageProfile.OsDisk != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet != nil &&
-				vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID != nil {
-				// set diskEncryptionSet as value of os disk by default
-				diskEncryptionSetID = *vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID
+
+	for k, v := range diskMap {
+		diskURI := k
+		opt := v
+		if opt.isManagedDisk {
+			attached := false
+			for _, disk := range *vm.StorageProfile.DataDisks {
+				if disk.ManagedDisk != nil && strings.EqualFold(*disk.ManagedDisk.ID, diskURI) {
+					attached = true
+					break
+				}
 			}
+			if attached {
+				klog.V(2).Infof("azureDisk - disk(%s) already attached to node(%s)", diskURI, nodeName)
+				continue
+			}
+
+			managedDisk := &compute.ManagedDiskParameters{ID: &diskURI}
+			if opt.diskEncryptionSetID == "" {
+				if vm.StorageProfile.OsDisk != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet != nil &&
+					vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID != nil {
+					// set diskEncryptionSet as value of os disk by default
+					opt.diskEncryptionSetID = *vm.StorageProfile.OsDisk.ManagedDisk.DiskEncryptionSet.ID
+				}
+			}
+			if opt.diskEncryptionSetID != "" {
+				managedDisk.DiskEncryptionSet = &compute.DiskEncryptionSetParameters{ID: &opt.diskEncryptionSetID}
+			}
+			disks = append(disks,
+				compute.DataDisk{
+					Name:                    &opt.diskName,
+					Lun:                     &opt.lun,
+					Caching:                 compute.CachingTypes(opt.cachingMode),
+					CreateOption:            "attach",
+					ManagedDisk:             managedDisk,
+					WriteAcceleratorEnabled: to.BoolPtr(opt.writeAcceleratorEnabled),
+				})
+		} else {
+			disks = append(disks,
+				compute.DataDisk{
+					Name: &opt.diskName,
+					Vhd: &compute.VirtualHardDisk{
+						URI: &diskURI,
+					},
+					Lun:          &opt.lun,
+					Caching:      compute.CachingTypes(opt.cachingMode),
+					CreateOption: "attach",
+				})
 		}
-		if diskEncryptionSetID != "" {
-			managedDisk.DiskEncryptionSet = &compute.DiskEncryptionSetParameters{ID: &diskEncryptionSetID}
-		}
-		disks = append(disks,
-			compute.DataDisk{
-				Name:                    &diskName,
-				Lun:                     &lun,
-				Caching:                 compute.CachingTypes(cachingMode),
-				CreateOption:            "attach",
-				ManagedDisk:             managedDisk,
-				WriteAcceleratorEnabled: to.BoolPtr(writeAcceleratorEnabled),
-			})
-	} else {
-		disks = append(disks,
-			compute.DataDisk{
-				Name: &diskName,
-				Vhd: &compute.VirtualHardDisk{
-					URI: &diskURI,
-				},
-				Lun:          &lun,
-				Caching:      compute.CachingTypes(cachingMode),
-				CreateOption: "attach",
-			})
 	}
+
 	newVM := compute.VirtualMachineScaleSetVM{
 		VirtualMachineScaleSetVMProperties: &compute.VirtualMachineScaleSetVMProperties{
 			StorageProfile: &compute.StorageProfile{
@@ -99,28 +116,27 @@ func (ss *scaleSet) AttachDisk(isManagedDisk bool, diskName, diskURI string, nod
 		_ = ss.deleteCacheForNode(vmName)
 	}()
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk(%s, %s) with DiskEncryptionSetID(%s)", nodeResourceGroup, nodeName, diskName, diskURI, diskEncryptionSetID)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk list(%s)", nodeResourceGroup, nodeName, diskMap)
 	rerr := ss.VirtualMachineScaleSetVMsClient.Update(ctx, nodeResourceGroup, ssName, instanceID, newVM, "attach_disk")
 	if rerr != nil {
-		klog.Errorf("azureDisk - attach disk(%s, %s) on rg(%s) vm(%s) failed, err: %v", diskName, diskURI, nodeResourceGroup, nodeName, rerr)
+		klog.Errorf("azureDisk - attach disk list(%s) on rg(%s) vm(%s) failed, err: %v", diskMap, nodeResourceGroup, nodeName, rerr)
 		if rerr.HTTPStatusCode == http.StatusNotFound {
-			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%s, %s) on rg(%s) vm(%s)", diskName, diskURI, nodeResourceGroup, nodeName)
+			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%v) on rg(%s) vm(%s)", diskMap, nodeResourceGroup, nodeName)
 			disks := ss.filterNonExistingDisks(ctx, *newVM.VirtualMachineScaleSetVMProperties.StorageProfile.DataDisks)
 			newVM.VirtualMachineScaleSetVMProperties.StorageProfile.DataDisks = &disks
 			rerr = ss.VirtualMachineScaleSetVMsClient.Update(ctx, nodeResourceGroup, ssName, instanceID, newVM, "attach_disk")
 		}
 	}
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk(%s, %s)  returned with %v", nodeResourceGroup, nodeName, diskName, diskURI, rerr)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - attach disk list(%s, %s) returned with %v", nodeResourceGroup, nodeName, diskMap, rerr)
 	if rerr != nil {
 		return rerr.Error()
 	}
 	return nil
 }
 
-// DetachDisk detaches a disk from host
-// the vhd can be identified by diskName or diskURI
-func (ss *scaleSet) DetachDisk(diskName, diskURI string, nodeName types.NodeName) error {
+// DetachDisk detaches a disk from VM
+func (ss *scaleSet) DetachDisk(nodeName types.NodeName, diskMap map[string]string) error {
 	vmName := mapNodeNameToVMName(nodeName)
 	ssName, instanceID, vm, err := ss.getVmssVM(vmName, azcache.CacheReadTypeDefault)
 	if err != nil {
@@ -139,24 +155,25 @@ func (ss *scaleSet) DetachDisk(diskName, diskURI string, nodeName types.NodeName
 	}
 	bFoundDisk := false
 	for i, disk := range disks {
-		if disk.Lun != nil && (disk.Name != nil && diskName != "" && strings.EqualFold(*disk.Name, diskName)) ||
-			(disk.Vhd != nil && disk.Vhd.URI != nil && diskURI != "" && strings.EqualFold(*disk.Vhd.URI, diskURI)) ||
-			(disk.ManagedDisk != nil && diskURI != "" && strings.EqualFold(*disk.ManagedDisk.ID, diskURI)) {
-			// found the disk
-			klog.V(2).Infof("azureDisk - detach disk: name %q uri %q", diskName, diskURI)
-			if strings.EqualFold(ss.cloud.Environment.Name, AzureStackCloudName) {
-				disks = append(disks[:i], disks[i+1:]...)
-			} else {
-				disks[i].ToBeDetached = to.BoolPtr(true)
+		for diskURI, diskName := range diskMap {
+			if disk.Lun != nil && (disk.Name != nil && diskName != "" && strings.EqualFold(*disk.Name, diskName)) ||
+				(disk.Vhd != nil && disk.Vhd.URI != nil && diskURI != "" && strings.EqualFold(*disk.Vhd.URI, diskURI)) ||
+				(disk.ManagedDisk != nil && diskURI != "" && strings.EqualFold(*disk.ManagedDisk.ID, diskURI)) {
+				// found the disk
+				klog.V(2).Infof("azureDisk - detach disk: name %q uri %q", diskName, diskURI)
+				if strings.EqualFold(ss.cloud.Environment.Name, AzureStackCloudName) {
+					disks = append(disks[:i], disks[i+1:]...)
+				} else {
+					disks[i].ToBeDetached = to.BoolPtr(true)
+				}
+				bFoundDisk = true
 			}
-			bFoundDisk = true
-			break
 		}
 	}
 
 	if !bFoundDisk {
 		// only log here, next action is to update VM status with original meta data
-		klog.Errorf("detach azure disk: disk %s not found, diskURI: %s", diskName, diskURI)
+		klog.Errorf("detach azure disk on node(%s): disk list(%s) not found", nodeName, diskMap)
 	}
 
 	newVM := compute.VirtualMachineScaleSetVM{
@@ -175,19 +192,19 @@ func (ss *scaleSet) DetachDisk(diskName, diskURI string, nodeName types.NodeName
 		_ = ss.deleteCacheForNode(vmName)
 	}()
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk(%s, %s)", nodeResourceGroup, nodeName, diskName, diskURI)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk list(%s)", nodeResourceGroup, nodeName, diskMap)
 	rerr := ss.VirtualMachineScaleSetVMsClient.Update(ctx, nodeResourceGroup, ssName, instanceID, newVM, "detach_disk")
 	if rerr != nil {
-		klog.Errorf("azureDisk - detach disk(%s, %s) on rg(%s) vm(%s) failed, err: %v", diskName, diskURI, nodeResourceGroup, nodeName, rerr)
+		klog.Errorf("azureDisk - detach disk list(%s) on rg(%s) vm(%s) failed, err: %v", diskMap, nodeResourceGroup, nodeName, rerr)
 		if rerr.HTTPStatusCode == http.StatusNotFound {
-			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%s, %s) on rg(%s) vm(%s)", diskName, diskURI, nodeResourceGroup, nodeName)
+			klog.Errorf("azureDisk - begin to filterNonExistingDisks(%v) on rg(%s) vm(%s)", diskMap, nodeResourceGroup, nodeName)
 			disks := ss.filterNonExistingDisks(ctx, *newVM.VirtualMachineScaleSetVMProperties.StorageProfile.DataDisks)
 			newVM.VirtualMachineScaleSetVMProperties.StorageProfile.DataDisks = &disks
 			rerr = ss.VirtualMachineScaleSetVMsClient.Update(ctx, nodeResourceGroup, ssName, instanceID, newVM, "detach_disk")
 		}
 	}
 
-	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk(%s, %s) returned with %v", nodeResourceGroup, nodeName, diskName, diskURI, rerr)
+	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - detach disk(%v) returned with %v", nodeResourceGroup, nodeName, diskMap, rerr)
 	if rerr != nil {
 		return rerr.Error()
 	}
diff --git a/pkg/provider/azure_managedDiskController.go b/pkg/provider/azure_managedDiskController.go
index a8dedbcf9..5f0264523 100644
--- a/pkg/provider/azure_managedDiskController.go
+++ b/pkg/provider/azure_managedDiskController.go
@@ -219,7 +219,7 @@ func (c *ManagedDiskController) DeleteManagedDisk(diskURI string) error {
 	ctx, cancel := getContextWithCancel()
 	defer cancel()
 
-	if _, ok := c.common.diskAttachDetachMap.Load(strings.ToLower(diskURI)); ok {
+	if _, ok := c.common.diskStateMap.Load(strings.ToLower(diskURI)); ok {
 		return fmt.Errorf("failed to delete disk(%s) since it's in attaching or detaching state", diskURI)
 	}
 
diff --git a/pkg/mockvmsets/azure_mock_vmsets.go b/pkg/provider/azure_mock_vmsets.go
similarity index 91%
rename from pkg/mockvmsets/azure_mock_vmsets.go
rename to pkg/provider/azure_mock_vmsets.go
index 14bf0637b..d4cb84c68 100644
--- a/pkg/mockvmsets/azure_mock_vmsets.go
+++ b/pkg/provider/azure_mock_vmsets.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mockvmsets
+package azure
 
 import (
 	reflect "reflect"
@@ -219,31 +219,31 @@ func (mr *MockVMSetMockRecorder) EnsureBackendPoolDeleted(service, backendPoolID
 }
 
 // AttachDisk mocks base method
-func (m *MockVMSet) AttachDisk(isManagedDisk bool, diskName, diskURI string, nodeName types.NodeName, lun int32, cachingMode compute.CachingTypes, diskEncryptionSetID string, writeAcceleratorEnabled bool) error {
+func (m *MockVMSet) AttachDisk(nodeName types.NodeName, diskMap map[string]*AttachDiskOptions) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "AttachDisk", isManagedDisk, diskName, diskURI, nodeName, lun, cachingMode, diskEncryptionSetID, writeAcceleratorEnabled)
+	ret := m.ctrl.Call(m, "AttachDisk", nodeName, diskMap)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // AttachDisk indicates an expected call of AttachDisk
-func (mr *MockVMSetMockRecorder) AttachDisk(isManagedDisk, diskName, diskURI, nodeName, lun, cachingMode, diskEncryptionSetID, writeAcceleratorEnabled interface{}) *gomock.Call {
+func (mr *MockVMSetMockRecorder) AttachDisk(nodeName, diskMap interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AttachDisk", reflect.TypeOf((*MockVMSet)(nil).AttachDisk), isManagedDisk, diskName, diskURI, nodeName, lun, cachingMode, diskEncryptionSetID, writeAcceleratorEnabled)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AttachDisk", reflect.TypeOf((*MockVMSet)(nil).AttachDisk), nodeName, diskMap)
 }
 
 // DetachDisk mocks base method
-func (m *MockVMSet) DetachDisk(diskName, diskURI string, nodeName types.NodeName) error {
+func (m *MockVMSet) DetachDisk(nodeName types.NodeName, diskMap map[string]string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DetachDisk", diskName, diskURI, nodeName)
+	ret := m.ctrl.Call(m, "DetachDisk", nodeName, diskMap)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // DetachDisk indicates an expected call of DetachDisk
-func (mr *MockVMSetMockRecorder) DetachDisk(diskName, diskURI, nodeName interface{}) *gomock.Call {
+func (mr *MockVMSetMockRecorder) DetachDisk(nodeName, diskMap interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DetachDisk", reflect.TypeOf((*MockVMSet)(nil).DetachDisk), diskName, diskURI, nodeName)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DetachDisk", reflect.TypeOf((*MockVMSet)(nil).DetachDisk), nodeName, diskMap)
 }
 
 // GetDataDisks mocks base method
diff --git a/pkg/provider/azure_vmsets.go b/pkg/provider/azure_vmsets.go
index 2023afa1a..4e01a5110 100644
--- a/pkg/provider/azure_vmsets.go
+++ b/pkg/provider/azure_vmsets.go
@@ -65,10 +65,10 @@ type VMSet interface {
 	// EnsureBackendPoolDeleted ensures the loadBalancer backendAddressPools deleted from the specified nodes.
 	EnsureBackendPoolDeleted(service *v1.Service, backendPoolID, vmSetName string, backendAddressPools *[]network.BackendAddressPool) error
 
-	// AttachDisk attaches a vhd to vm. The vhd must exist, can be identified by diskName, diskURI, and lun.
-	AttachDisk(isManagedDisk bool, diskName, diskURI string, nodeName types.NodeName, lun int32, cachingMode compute.CachingTypes, diskEncryptionSetID string, writeAcceleratorEnabled bool) error
-	// DetachDisk detaches a vhd from host. The vhd can be identified by diskName or diskURI.
-	DetachDisk(diskName, diskURI string, nodeName types.NodeName) error
+	// AttachDisk attaches a disk to vm
+	AttachDisk(nodeName types.NodeName, diskMap map[string]*AttachDiskOptions) error
+	// DetachDisk detaches a disk from vm
+	DetachDisk(nodeName types.NodeName, diskMap map[string]string) error
 	// GetDataDisks gets a list of data disks attached to the node.
 	GetDataDisks(nodeName types.NodeName, string azcache.AzureCacheReadType) ([]compute.DataDisk, error)
 


Test patch:
diff --git a/pkg/provider/azure_controller_common_test.go b/pkg/provider/azure_controller_common_test.go
index 324fee46f..11b215887 100644
--- a/pkg/provider/azure_controller_common_test.go
+++ b/pkg/provider/azure_controller_common_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net/http"
 	"reflect"
+	"strings"
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-06-30/compute"
@@ -131,7 +132,7 @@ func TestCommonAttachDisk(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
 		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
 			testCloud.SubscriptionID, testCloud.ResourceGroup, *test.existedDisk.Name)
@@ -235,7 +236,7 @@ func TestCommonAttachDiskWithVMSS(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
 		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
 			testCloud.SubscriptionID, testCloud.ResourceGroup, *test.existedDisk.Name)
@@ -309,7 +310,7 @@ func TestCommonDetachDisk(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
 		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/disk-name",
 			testCloud.SubscriptionID, testCloud.ResourceGroup)
@@ -367,7 +368,7 @@ func TestGetDiskLun(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
 		expectedVMs := setTestVirtualMachines(testCloud, map[string]string{"vm1": "PowerState/Running"}, false)
 		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
@@ -381,28 +382,44 @@ func TestGetDiskLun(t *testing.T) {
 	}
 }
 
-func TestGetNextDiskLun(t *testing.T) {
+func TestSetDiskLun(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
 	testCases := []struct {
 		desc            string
+		nodeName        string
+		diskURI         string
+		diskMap         map[string]*AttachDiskOptions
 		isDataDisksFull bool
 		expectedLun     int32
 		expectedErr     bool
 	}{
 		{
-			desc:            "the minimal LUN shall be returned if there's enough room for extra disks",
-			isDataDisksFull: false,
-			expectedLun:     1,
-			expectedErr:     false,
+			desc:        "the minimal LUN shall be returned if there's enough room for extra disks",
+			nodeName:    "nodeName",
+			diskURI:     "diskURI",
+			diskMap:     map[string]*AttachDiskOptions{"diskURI": {}},
+			expectedLun: 1,
+			expectedErr: false,
 		},
 		{
-			desc:            "LUN -1 and  error shall be returned if there's no available LUN",
+			desc:            "LUN -1 and error shall be returned if there's no available LUN",
+			nodeName:        "nodeName",
+			diskURI:         "diskURI",
+			diskMap:         map[string]*AttachDiskOptions{"diskURI": {}},
 			isDataDisksFull: true,
 			expectedLun:     -1,
 			expectedErr:     true,
 		},
+		{
+			desc:        "diskURI1 is not in VM data disk list nor in diskMap",
+			nodeName:    "nodeName",
+			diskURI:     "diskURI1",
+			diskMap:     map[string]*AttachDiskOptions{"diskURI2": {}},
+			expectedLun: -1,
+			expectedErr: true,
+		},
 	}
 
 	for i, test := range testCases {
@@ -413,15 +430,15 @@ func TestGetNextDiskLun(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
-		expectedVMs := setTestVirtualMachines(testCloud, map[string]string{"vm1": "PowerState/Running"}, test.isDataDisksFull)
+		expectedVMs := setTestVirtualMachines(testCloud, map[string]string{test.nodeName: "PowerState/Running"}, test.isDataDisksFull)
 		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
 		for _, vm := range expectedVMs {
 			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, *vm.Name, gomock.Any()).Return(vm, nil).AnyTimes()
 		}
 
-		lun, err := common.GetNextDiskLun("vm1")
+		lun, err := common.SetDiskLun(types.NodeName(test.nodeName), test.diskURI, test.diskMap)
 		assert.Equal(t, test.expectedLun, lun, "TestCase[%d]: %s", i, test.desc)
 		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
 	}
@@ -462,7 +479,7 @@ func TestDisksAreAttached(t *testing.T) {
 			resourceGroup:         testCloud.ResourceGroup,
 			subscriptionID:        testCloud.SubscriptionID,
 			cloud:                 testCloud,
-			vmLockMap:             newLockMap(),
+			lockMap:               newLockMap(),
 		}
 		expectedVMs := setTestVirtualMachines(testCloud, map[string]string{"vm1": "PowerState/Running"}, false)
 		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
@@ -652,7 +669,7 @@ func TestCheckDiskExists(t *testing.T) {
 		resourceGroup:         testCloud.ResourceGroup,
 		subscriptionID:        testCloud.SubscriptionID,
 		cloud:                 testCloud,
-		vmLockMap:             newLockMap(),
+		lockMap:               newLockMap(),
 	}
 	// create a new disk before running test
 	newDiskName := "newdisk"
@@ -706,7 +723,7 @@ func TestFilterNonExistingDisks(t *testing.T) {
 		resourceGroup:         testCloud.ResourceGroup,
 		subscriptionID:        testCloud.SubscriptionID,
 		cloud:                 testCloud,
-		vmLockMap:             newLockMap(),
+		lockMap:               newLockMap(),
 	}
 	// create a new disk before running test
 	diskURIPrefix := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/",
@@ -768,7 +785,7 @@ func TestFilterNonExistingDisksWithSpecialHTTPStatusCode(t *testing.T) {
 		resourceGroup:         testCloud.ResourceGroup,
 		subscriptionID:        testCloud.SubscriptionID,
 		cloud:                 testCloud,
-		vmLockMap:             newLockMap(),
+		lockMap:               newLockMap(),
 	}
 	// create a new disk before running test
 	diskURIPrefix := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/",
@@ -821,3 +838,162 @@ func TestIsInstanceNotFoundError(t *testing.T) {
 		assert.Equal(t, test.expectedResult, result, "TestCase[%d]", i, result)
 	}
 }
+
+func TestAttachDiskRequestFuncs(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc                 string
+		diskURI              string
+		nodeName             string
+		diskName             string
+		duplicateDiskRequest bool
+		diskNum              int
+		expectedErr          bool
+	}{
+		{
+			desc:        "one disk request in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     1,
+			expectedErr: false,
+		},
+		{
+			desc:        "multiple disk requests in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     10,
+			expectedErr: false,
+		},
+		{
+			desc:        "zero disk request in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     0,
+			expectedErr: false,
+		},
+		{
+			desc:                 "multiple disk requests in queue",
+			diskURI:              "diskURI",
+			nodeName:             "nodeName",
+			diskName:             "diskName",
+			duplicateDiskRequest: true,
+			diskNum:              10,
+			expectedErr:          false,
+		},
+	}
+
+	for i, test := range testCases {
+		testCloud := GetTestCloud(ctrl)
+		common := &controllerCommon{
+			location:              testCloud.Location,
+			storageEndpointSuffix: testCloud.Environment.StorageEndpointSuffix,
+			resourceGroup:         testCloud.ResourceGroup,
+			subscriptionID:        testCloud.SubscriptionID,
+			cloud:                 testCloud,
+			lockMap:               newLockMap(),
+		}
+		for i := 1; i <= test.diskNum; i++ {
+			diskURI := fmt.Sprintf("%s%d", test.diskURI, i)
+			diskName := fmt.Sprintf("%s%d", test.diskName, i)
+			attachDiskOptions := &AttachDiskOptions{diskName: diskName}
+			err := common.insertAttachDiskRequest(diskURI, test.nodeName, attachDiskOptions)
+			assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+			if test.duplicateDiskRequest {
+				err := common.insertAttachDiskRequest(diskURI, test.nodeName, attachDiskOptions)
+				assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+			}
+		}
+
+		diskMap, err := common.cleanAttachDiskRequests(test.nodeName)
+		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+		assert.Equal(t, test.diskNum, len(diskMap), "TestCase[%d]: %s", i, test.desc)
+		for diskURI, opt := range diskMap {
+			assert.Equal(t, strings.Contains(diskURI, test.diskURI), true, "TestCase[%d]: %s", i, test.desc)
+			assert.Equal(t, strings.Contains(opt.diskName, test.diskName), true, "TestCase[%d]: %s", i, test.desc)
+		}
+	}
+}
+
+func TestDetachDiskRequestFuncs(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc                 string
+		diskURI              string
+		nodeName             string
+		diskName             string
+		duplicateDiskRequest bool
+		diskNum              int
+		expectedErr          bool
+	}{
+		{
+			desc:        "one disk request in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     1,
+			expectedErr: false,
+		},
+		{
+			desc:        "multiple disk requests in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     10,
+			expectedErr: false,
+		},
+		{
+			desc:        "zero disk request in queue",
+			diskURI:     "diskURI",
+			nodeName:    "nodeName",
+			diskName:    "diskName",
+			diskNum:     0,
+			expectedErr: false,
+		},
+		{
+			desc:                 "multiple disk requests in queue",
+			diskURI:              "diskURI",
+			nodeName:             "nodeName",
+			diskName:             "diskName",
+			duplicateDiskRequest: true,
+			diskNum:              10,
+			expectedErr:          false,
+		},
+	}
+
+	for i, test := range testCases {
+		testCloud := GetTestCloud(ctrl)
+		common := &controllerCommon{
+			location:              testCloud.Location,
+			storageEndpointSuffix: testCloud.Environment.StorageEndpointSuffix,
+			resourceGroup:         testCloud.ResourceGroup,
+			subscriptionID:        testCloud.SubscriptionID,
+			cloud:                 testCloud,
+			lockMap:               newLockMap(),
+		}
+		for i := 1; i <= test.diskNum; i++ {
+			diskURI := fmt.Sprintf("%s%d", test.diskURI, i)
+			diskName := fmt.Sprintf("%s%d", test.diskName, i)
+			err := common.insertDetachDiskRequest(diskName, diskURI, test.nodeName)
+			assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+			if test.duplicateDiskRequest {
+				err := common.insertDetachDiskRequest(diskName, diskURI, test.nodeName)
+				assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+			}
+		}
+
+		diskMap, err := common.cleanDetachDiskRequests(test.nodeName)
+		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s", i, test.desc)
+		assert.Equal(t, test.diskNum, len(diskMap), "TestCase[%d]: %s", i, test.desc)
+		for diskURI, diskName := range diskMap {
+			assert.Equal(t, strings.Contains(diskURI, test.diskURI), true, "TestCase[%d]: %s", i, test.desc)
+			assert.Equal(t, strings.Contains(diskName, test.diskName), true, "TestCase[%d]: %s", i, test.desc)
+		}
+	}
+}
diff --git a/pkg/provider/azure_controller_standard_test.go b/pkg/provider/azure_controller_standard_test.go
index d531f4a61..7209ff730 100644
--- a/pkg/provider/azure_controller_standard_test.go
+++ b/pkg/provider/azure_controller_standard_test.go
@@ -103,8 +103,18 @@ func TestStandardAttachDisk(t *testing.T) {
 			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		}
 
-		err := vmSet.AttachDisk(test.isManagedDisk, "",
-			"uri", test.nodeName, 0, compute.CachingTypesReadOnly, "", false)
+		options := AttachDiskOptions{
+			lun:                     0,
+			isManagedDisk:           test.isManagedDisk,
+			diskName:                "",
+			cachingMode:             compute.CachingTypesReadOnly,
+			diskEncryptionSetID:     "",
+			writeAcceleratorEnabled: false,
+		}
+		diskMap := map[string]*AttachDiskOptions{
+			"uri": &options,
+		}
+		err := vmSet.AttachDisk(test.nodeName, diskMap)
 		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s, err: %v", i, test.desc, err)
 	}
 }
@@ -160,7 +170,10 @@ func TestStandardDetachDisk(t *testing.T) {
 			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		}
 
-		err := vmSet.DetachDisk(test.diskName, "", test.nodeName)
+		diskMap := map[string]string{
+			"uri": test.diskName,
+		}
+		err := vmSet.DetachDisk(test.nodeName, diskMap)
 		assert.Equal(t, test.expectedError, err != nil, "TestCase[%d]: %s", i, test.desc)
 		if !test.expectedError && test.diskName != "" {
 			dataDisks, err := vmSet.GetDataDisks(test.nodeName, azcache.CacheReadTypeDefault)
diff --git a/pkg/provider/azure_controller_vmss_test.go b/pkg/provider/azure_controller_vmss_test.go
index 96f0a4d34..81fa63d97 100644
--- a/pkg/provider/azure_controller_vmss_test.go
+++ b/pkg/provider/azure_controller_vmss_test.go
@@ -129,7 +129,18 @@ func TestAttachDiskWithVMSS(t *testing.T) {
 		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
 			testCloud.SubscriptionID, testCloud.ResourceGroup, *test.existedDisk.Name)
 
-		err = ss.AttachDisk(test.isManagedDisk, "disk-name", diskURI, test.vmssvmName, 0, compute.CachingTypesReadWrite, "", true)
+		options := AttachDiskOptions{
+			lun:                     0,
+			isManagedDisk:           test.isManagedDisk,
+			diskName:                "disk-name",
+			cachingMode:             compute.CachingTypesReadWrite,
+			diskEncryptionSetID:     "",
+			writeAcceleratorEnabled: true,
+		}
+		diskMap := map[string]*AttachDiskOptions{
+			diskURI: &options,
+		}
+		err = ss.AttachDisk(test.vmssvmName, diskMap)
 		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s, return error: %v", i, test.desc, err)
 		assert.Equal(t, test.expectedErrMsg, err, "TestCase[%d]: %s, expected error: %v, return error: %v", i, test.desc, test.expectedErrMsg, err)
 	}
@@ -225,7 +236,10 @@ func TestDetachDiskWithVMSS(t *testing.T) {
 			mockVMSSVMClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, scaleSetName, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		}
 
-		err = ss.DetachDisk(*test.existedDisk.Name, diskName, test.vmssvmName)
+		diskMap := map[string]string{
+			diskName: *test.existedDisk.Name,
+		}
+		err = ss.DetachDisk(test.vmssvmName, diskMap)
 		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s, err: %v", i, test.desc, err)
 		assert.Equal(t, test.expectedErrMsg, err, "TestCase[%d]: %s, expected error: %v, return error: %v", i, test.desc, test.expectedErrMsg, err)
 
diff --git a/pkg/provider/azure_managedDiskController_test.go b/pkg/provider/azure_managedDiskController_test.go
index 82ffba3e6..34c5bbe6a 100644
--- a/pkg/provider/azure_managedDiskController_test.go
+++ b/pkg/provider/azure_managedDiskController_test.go
@@ -218,7 +218,7 @@ func TestDeleteManagedDisk(t *testing.T) {
 		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
 			testCloud.SubscriptionID, testCloud.ResourceGroup, *test.existedDisk.Name)
 		if test.diskState == "attaching" {
-			managedDiskController.common.diskAttachDetachMap.Store(strings.ToLower(diskURI), test.diskState)
+			managedDiskController.common.diskStateMap.Store(strings.ToLower(diskURI), test.diskState)
 		}
 
 		mockDisksClient := testCloud.DisksClient.(*mockdiskclient.MockInterface)
diff --git a/pkg/provider/azure_routes_test.go b/pkg/provider/azure_routes_test.go
index 6b8278f5e..56f535001 100644
--- a/pkg/provider/azure_routes_test.go
+++ b/pkg/provider/azure_routes_test.go
@@ -34,7 +34,6 @@ import (
 	cloudprovider "k8s.io/cloud-provider"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/routetableclient/mockroutetableclient"
-	"sigs.k8s.io/cloud-provider-azure/pkg/mockvmsets"
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
@@ -114,7 +113,7 @@ func TestCreateRoute(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	routeTableClient := mockroutetableclient.NewMockInterface(ctrl)
-	mockVMSet := mockvmsets.NewMockVMSet(ctrl)
+	mockVMSet := NewMockVMSet(ctrl)
 
 	cloud := &Cloud{
 		RouteTablesClient: routeTableClient,
@@ -521,7 +520,7 @@ func TestListRoutes(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	routeTableClient := mockroutetableclient.NewMockInterface(ctrl)
-	mockVMSet := mockvmsets.NewMockVMSet(ctrl)
+	mockVMSet := NewMockVMSet(ctrl)
 
 	cloud := &Cloud{
 		RouteTablesClient: routeTableClient,