Repository: kubernetes-sigs/cloud-provider-azure
Pull request #2791 (2022-11-17 23:12:55)

feat: detect errors that requires vm udpate retries and retry if detected
<!--  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/first-contribution.md#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/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/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. 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 cleanup

<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature
/kind design

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:
* This PR allows `cloud-provider-azure` to retry vm update on `OperationPreempted` error with http status code of 2xx, where storage config change has been committed but vm update has failed.
* This PR adds `PartialUpdateError` type to implement `error` interface to be returned for other VM update failures with status code of 2xx.
* This PR purports to address a reported customer issue where attaches fail due to `OperationPreempted` error.

#### 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:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
NONE
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


k8s-ci-robot: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
k8s-ci-robot: @sunpa93: Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our [release note process](https://git.k8s.io/community/contributors/guide/release-notes.md) to remove it.

<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.
</details>
netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *kubernetes-sigs-cloud-provide-azure* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 8a79b08c6dfbc07175cb27239b86a1b1f480093e |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/kubernetes-sigs-cloud-provide-azure/deploys/6376bff9613d4f00088e9f46 |
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/2791#" title="Author self-approved">sunpa93</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign feiskyer for approval by writing `/assign @feiskyer` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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).

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

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

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":["feiskyer"]} -->
coveralls: 
[![Coverage Status](https://coveralls.io/builds/54342276/badge)](https://coveralls.io/builds/54342276)

Coverage decreased (-0.2%) to 79.461% when pulling **8a79b08c6dfbc07175cb27239b86a1b1f480093e on sunpa93:handle-preemption-v1** into **a454d33d909086f46c99b449dc8da7cdcf406ee0 on kubernetes-sigs:master**.

sunpa93: e2e tests triggered for v1 driver to verify changes (https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1606)
k8s-ci-robot: @sunpa93: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-check | 8a79b08c6dfbc07175cb27239b86a1b1f480093e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2791/pull-cloud-provider-azure-check/1593759683556937728) | true | `/test pull-cloud-provider-azure-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2791). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asunpa93). 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 -->
sunpa93: @feiskyer Could you review the PR? The change is to address a reported customer issue with `OperationPreempted` error. The change has been tested on both v1 (https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1606) and v2 driver (https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1605)
andyzhangx: > @feiskyer Could you review the PR? The change is to address a reported customer issue with `OperationPreempted` error. The change has been tested on both v1 ([kubernetes-sigs/azuredisk-csi-driver#1606](https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1606)) and v2 driver ([kubernetes-sigs/azuredisk-csi-driver#1605](https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1605))

@sunpa93 sorry for the late review, just ping me when you have any other questions, thanks.
k8s-ci-robot: @sunpa93: PR needs rebase.

<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.
</details>
k8s-ci-robot: @sunpa93: PR needs rebase.

<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.
</details>
k8s-ci-robot: @sunpa93: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-unit | 3ae53c5176adb58ce16bde9b4ece60e8654fdd97 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/2791/pull-cloud-provider-azure-unit/1602475627250716672) | true | `/test pull-cloud-provider-azure-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=2791). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asunpa93). 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 -->
sunpa93: /retest
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/2791#pullrequestreview-1215021313" title="Approved">andyzhangx</a>*, *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/2791#" title="Author self-approved">sunpa93</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
sunpa93: tracking e2e tests with the change: https://github.com/kubernetes-sigs/azuredisk-csi-driver/pull/1606

Patch:
diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go
index 04838b5d1..80d26c353 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -383,6 +383,8 @@ const (
 	CannotUpdateVMBeingDeletedMessagePrefix = "'Put on Virtual Machine Scale Set VM Instance' is not allowed on Virtual Machine Scale Set"
 	// CannotUpdateVMBeingDeletedMessageSuffix is the suffix of the error message that the request failed due to delete a VM that is being deleted
 	CannotUpdateVMBeingDeletedMessageSuffix = "since it is marked for deletion"
+	// OperationPreemptedErrorCode is the error code returned for vm operation preempted errors
+	OperationPreemptedErrorCode = "OperationPreempted"
 )
 
 // node ipam controller
diff --git a/pkg/nodeipam/node_ipam_controller.go b/pkg/nodeipam/node_ipam_controller.go
index b31aa5ed7..82e49ca77 100644
--- a/pkg/nodeipam/node_ipam_controller.go
+++ b/pkg/nodeipam/node_ipam_controller.go
@@ -80,7 +80,7 @@ func registerRateLimiterMetric(ownerName string) error {
 		StabilityLevel: metrics.ALPHA,
 	})
 	if err := legacyregistry.Register(metric); err != nil {
-		return fmt.Errorf("error registering rate limiter usage metric: %v", err)
+		return fmt.Errorf("error registering rate limiter usage metric: %w", err)
 	}
 	stopCh := make(chan struct{})
 	rateLimiterMetrics[ownerName] = &rateLimiterMetric{
diff --git a/pkg/provider/azure_controller_common.go b/pkg/provider/azure_controller_common.go
index 883e30e0c..149282324 100644
--- a/pkg/provider/azure_controller_common.go
+++ b/pkg/provider/azure_controller_common.go
@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-03-01/compute"
+	"github.com/Azure/go-autorest/autorest/azure"
 
 	"k8s.io/apimachinery/pkg/types"
 	kwait "k8s.io/apimachinery/pkg/util/wait"
@@ -38,6 +39,7 @@ import (
 
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
 const (
@@ -54,6 +56,10 @@ const (
 	attachDiskMapKeySuffix = "attachdiskmap"
 	detachDiskMapKeySuffix = "detachdiskmap"
 
+	updateVMRetryDuration = time.Duration(1) * time.Second
+	updateVMRetryFactor   = 3.0
+	updateVMRetrySteps    = 5
+
 	// WriteAcceleratorEnabled support for Azure Write Accelerator on Azure Disks
 	// https://docs.microsoft.com/azure/virtual-machines/windows/how-to-enable-write-accelerator
 	WriteAcceleratorEnabled = "writeacceleratorenabled"
@@ -72,9 +78,16 @@ var defaultBackOff = kwait.Backoff{
 	Jitter:   0.0,
 }
 
+var updateVMBackoff = kwait.Backoff{
+	Duration: updateVMRetryDuration,
+	Factor:   updateVMRetryFactor,
+	Steps:    updateVMRetrySteps,
+}
+
 var (
 	managedDiskPathRE  = regexp.MustCompile(`.*/subscriptions/(?:.*)/resourceGroups/(?:.*)/providers/Microsoft.Compute/disks/(.+)`)
 	diskSnapshotPathRE = regexp.MustCompile(`.*/subscriptions/(?:.*)/resourceGroups/(?:.*)/providers/Microsoft.Compute/snapshots/(.+)`)
+	errorCodeRE        = regexp.MustCompile(`Code="(.*?)".*`)
 )
 
 type controllerCommon struct {
@@ -254,9 +267,10 @@ func (c *controllerCommon) AttachDisk(ctx context.Context, async bool, diskName,
 	c.diskStateMap.Store(disk, "attaching")
 	defer c.diskStateMap.Delete(disk)
 	future, err := vmset.AttachDisk(ctx, nodeName, diskMap)
-	if err != nil {
-		return -1, err
+	if future == nil {
+		return -1, fmt.Errorf("nil future was returned: %w", err)
 	}
+	// err will be handled by waitForUpdateResult below
 
 	if async && c.diskOpRateLimiter.TryAccept() {
 		// unlock and wait for attach disk complete
@@ -267,7 +281,40 @@ func (c *controllerCommon) AttachDisk(ctx context.Context, async bool, diskName,
 			klog.Warningf("azureDisk - switch to batch operation due to rate limited, QPS: %f", c.diskOpRateLimiter.QPS())
 		}
 	}
-	return lun, vmset.WaitForUpdateResult(ctx, future, nodeName, "attach_disk")
+
+	if err = c.waitForUpdateResult(ctx, vmset, nodeName, future, err); err != nil {
+		return -1, err
+	}
+	return lun, nil
+}
+
+// waitForUpdateResult handles asynchronous VM update operations and retries with backoff if OperationPreempted error is observed
+func (c *controllerCommon) waitForUpdateResult(ctx context.Context, vmset VMSet, nodeName types.NodeName, future *azure.Future, updateErr error) (err error) {
+	err = updateErr
+
+	if err == nil {
+		err = vmset.WaitForUpdateResult(ctx, future, nodeName, "attach_disk")
+	}
+
+	if vmUpdateRequired(future, err) {
+		if derr := kwait.ExponentialBackoffWithContext(ctx, updateVMBackoff, func() (bool, error) {
+			klog.Errorf("Retry VM Update on node (%s) due to error (%v)", nodeName, err)
+			future, err = vmset.UpdateVMAsync(ctx, nodeName)
+			if err == nil {
+				err = vmset.WaitForUpdateResult(ctx, future, nodeName, "attach_disk")
+			}
+			return !vmUpdateRequired(future, err), nil
+		}); derr != nil {
+			err = derr
+			return
+		}
+	}
+
+	if err != nil && configAccepted(future) {
+		err = retry.NewPartialUpdateError(err.Error())
+	}
+
+	return
 }
 
 func (c *controllerCommon) insertAttachDiskRequest(diskURI, nodeName string, options *AttachDiskOptions) error {
@@ -620,6 +667,11 @@ func (c *controllerCommon) checkDiskExists(ctx context.Context, diskURI string)
 	return true, nil
 }
 
+func vmUpdateRequired(future *azure.Future, err error) bool {
+	errCode := getAzureErrorCode(err)
+	return configAccepted(future) && errCode == consts.OperationPreemptedErrorCode
+}
+
 func getValidCreationData(subscriptionID, resourceGroup, sourceResourceID, sourceType string) (compute.CreationData, error) {
 	if sourceResourceID == "" {
 		return compute.CreationData{
@@ -663,3 +715,21 @@ func isInstanceNotFoundError(err error) bool {
 	}
 	return strings.Contains(errMsg, errStatusCode400) && strings.Contains(errMsg, errInvalidParameter) && strings.Contains(errMsg, errTargetInstanceIds)
 }
+
+// getAzureErrorCode uses regex to parse out the error code encapsulated in the error string.
+func getAzureErrorCode(err error) string {
+	if err == nil {
+		return ""
+	}
+	matches := errorCodeRE.FindStringSubmatch(err.Error())
+	if matches == nil {
+		return ""
+	}
+	return matches[1]
+}
+
+// configAccepted returns true if storage profile change had been committed (i.e. HTTP status code == 2xx) and returns false otherwise.
+func configAccepted(future *azure.Future) bool {
+	// if status code indicates success, the storage profile change was committed
+	return future != nil && future.Response() != nil && future.Response().StatusCode/100 == 2
+}
diff --git a/pkg/provider/azure_controller_standard.go b/pkg/provider/azure_controller_standard.go
index 16ae72ec9..c67172eb4 100644
--- a/pkg/provider/azure_controller_standard.go
+++ b/pkg/provider/azure_controller_standard.go
@@ -223,10 +223,20 @@ func (as *availabilitySet) DetachDisk(ctx context.Context, nodeName types.NodeNa
 
 // UpdateVM updates a vm
 func (as *availabilitySet) UpdateVM(ctx context.Context, nodeName types.NodeName) error {
+	future, err := as.UpdateVMAsync(ctx, nodeName)
+	if err != nil {
+		return err
+	}
+
+	return as.WaitForUpdateResult(ctx, future, nodeName, "update_vm")
+}
+
+// UpdateVMAsync updates a vm asynchronously
+func (as *availabilitySet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error) {
 	vmName := mapNodeNameToVMName(nodeName)
 	nodeResourceGroup, err := as.GetNodeResourceGroup(vmName)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s)", nodeResourceGroup, vmName)
 	// Invalidate the cache right after updating
@@ -234,12 +244,12 @@ func (as *availabilitySet) UpdateVM(ctx context.Context, nodeName types.NodeName
 		_ = as.DeleteCacheForNode(vmName)
 	}()
 
-	rerr := as.VirtualMachinesClient.Update(ctx, nodeResourceGroup, vmName, compute.VirtualMachineUpdate{}, "update_vm")
+	future, rerr := as.VirtualMachinesClient.UpdateAsync(ctx, nodeResourceGroup, vmName, compute.VirtualMachineUpdate{}, "update_vm")
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - returned with %v", nodeResourceGroup, vmName, rerr)
 	if rerr != nil {
-		return rerr.Error()
+		return future, rerr.Error()
 	}
-	return nil
+	return future, nil
 }
 
 // GetDataDisks gets a list of data disks attached to the node.
diff --git a/pkg/provider/azure_controller_vmss.go b/pkg/provider/azure_controller_vmss.go
index 922bae588..4178913f7 100644
--- a/pkg/provider/azure_controller_vmss.go
+++ b/pkg/provider/azure_controller_vmss.go
@@ -263,20 +263,31 @@ func (ss *ScaleSet) DetachDisk(ctx context.Context, nodeName types.NodeName, dis
 
 // UpdateVM updates a vm
 func (ss *ScaleSet) UpdateVM(ctx context.Context, nodeName types.NodeName) error {
+	future, err := ss.UpdateVMAsync(ctx, nodeName)
+	if err != nil {
+		return err
+	}
+
+	return ss.WaitForUpdateResult(ctx, future, nodeName, "update_vm")
+}
+
+// UpdateVMAsync updates a vm asynchronously
+func (ss *ScaleSet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error) {
 	vmName := mapNodeNameToVMName(nodeName)
 	vm, err := ss.getVmssVM(vmName, azcache.CacheReadTypeDefault)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	nodeResourceGroup, err := ss.GetNodeResourceGroup(vmName)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	var updateResult *compute.VirtualMachineScaleSetVM
+	var future *azure.Future
 	var rerr *retry.Error
 
+	// Invalidate the cache right after updating
 	defer func() {
 		// If there is an error with Update operation,
 		// invalidate the cache
@@ -284,26 +295,16 @@ func (ss *ScaleSet) UpdateVM(ctx context.Context, nodeName types.NodeName) error
 			_ = ss.DeleteCacheForNode(vmName)
 			return
 		}
-
-		// Update the cache with the updated result only if its not nil
-		// and contains the VirtualMachineScaleSetVMProperties
-		if updateResult != nil && updateResult.VirtualMachineScaleSetVMProperties != nil {
-			if err := ss.updateCache(vmName, nodeResourceGroup, vm.VMSSName, vm.InstanceID, updateResult); err != nil {
-				klog.Errorf("updateCache(%s, %s, %s) failed with error: %v", vmName, nodeResourceGroup, vm.VMSSName, vm.InstanceID, err)
-				// if err faced during updating cache, invalidate the cache
-				_ = ss.DeleteCacheForNode(vmName)
-			}
-		}
 	}()
 
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s)", nodeResourceGroup, nodeName)
-	updateResult, rerr = ss.VirtualMachineScaleSetVMsClient.Update(ctx, nodeResourceGroup, vm.VMSSName, vm.InstanceID, compute.VirtualMachineScaleSetVM{}, "update_vmss_instance")
+	future, rerr = ss.VirtualMachineScaleSetVMsClient.UpdateAsync(ctx, nodeResourceGroup, vm.VMSSName, vm.InstanceID, compute.VirtualMachineScaleSetVM{}, "update_vmss_instance")
 
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - returned with %v", nodeResourceGroup, nodeName, rerr)
 	if rerr != nil {
-		return rerr.Error()
+		return future, rerr.Error()
 	}
-	return nil
+	return future, nil
 }
 
 // GetDataDisks gets a list of data disks attached to the node.
diff --git a/pkg/provider/azure_controller_vmssflex.go b/pkg/provider/azure_controller_vmssflex.go
index e8cab8553..d526f4719 100644
--- a/pkg/provider/azure_controller_vmssflex.go
+++ b/pkg/provider/azure_controller_vmssflex.go
@@ -218,16 +218,27 @@ func (fs *FlexScaleSet) WaitForUpdateResult(ctx context.Context, future *azure.F
 
 // UpdateVM updates a vm
 func (fs *FlexScaleSet) UpdateVM(ctx context.Context, nodeName types.NodeName) error {
+	future, err := fs.UpdateVMAsync(ctx, nodeName)
+	if err != nil {
+		return err
+	}
+
+	return fs.WaitForUpdateResult(ctx, future, nodeName, "update_vm")
+}
+
+// UpdateVMAsync updates a vm asynchronously
+func (fs *FlexScaleSet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error) {
 	vmName := mapNodeNameToVMName(nodeName)
+
 	vm, err := fs.getVmssFlexVM(vmName, azcache.CacheReadTypeDefault)
 	if err != nil {
 		// if host doesn't exist, no need to update
 		klog.Warningf("azureDisk - cannot find node %s, skip updating vm)", nodeName)
-		return nil
+		return nil, nil
 	}
 	nodeResourceGroup, err := fs.GetNodeResourceGroup(vmName)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	defer func() {
@@ -236,12 +247,12 @@ func (fs *FlexScaleSet) UpdateVM(ctx context.Context, nodeName types.NodeName) e
 
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s)", nodeResourceGroup, vmName)
 
-	rerr := fs.VirtualMachinesClient.Update(ctx, nodeResourceGroup, *vm.Name, compute.VirtualMachineUpdate{}, "update_vm")
+	future, rerr := fs.VirtualMachinesClient.UpdateAsync(ctx, nodeResourceGroup, *vm.Name, compute.VirtualMachineUpdate{}, "update_vm")
 	klog.V(2).Infof("azureDisk - update(%s): vm(%s) - returned with %v", nodeResourceGroup, vmName, rerr)
 	if rerr != nil {
-		return rerr.Error()
+		return future, rerr.Error()
 	}
-	return nil
+	return future, nil
 }
 
 // GetDataDisks gets a list of data disks attached to the node.
diff --git a/pkg/provider/azure_mock_vmsets.go b/pkg/provider/azure_mock_vmsets.go
index be51095b0..ee14278f3 100644
--- a/pkg/provider/azure_mock_vmsets.go
+++ b/pkg/provider/azure_mock_vmsets.go
@@ -415,6 +415,21 @@ func (mr *MockVMSetMockRecorder) UpdateVM(ctx, nodeName interface{}) *gomock.Cal
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVM", reflect.TypeOf((*MockVMSet)(nil).UpdateVM), ctx, nodeName)
 }
 
+// UpdateVMAsync mocks base method
+func (m *MockVMSet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UpdateVMAsync", ctx, nodeName)
+	ret0, _ := ret[0].(*azure.Future)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// UpdateVMAsync indicates an expected call of UpdateVMAsync
+func (mr *MockVMSetMockRecorder) UpdateVMAsync(ctx, nodeName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVMAsync", reflect.TypeOf((*MockVMSet)(nil).UpdateVMAsync), ctx, nodeName)
+}
+
 // WaitForUpdateResult mocks base method.
 func (m *MockVMSet) WaitForUpdateResult(ctx context.Context, future *azure.Future, nodeName types.NodeName, source string) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/provider/azure_vmsets.go b/pkg/provider/azure_vmsets.go
index d82958796..e2935ffdc 100644
--- a/pkg/provider/azure_vmsets.go
+++ b/pkg/provider/azure_vmsets.go
@@ -88,6 +88,9 @@ type VMSet interface {
 	// UpdateVM updates a vm
 	UpdateVM(ctx context.Context, nodeName types.NodeName) error
 
+	// UpdateVMAsync updates a vm asynchronously
+	UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error)
+
 	// GetPowerStatusByNodeName returns the powerState for the specified node.
 	GetPowerStatusByNodeName(name string) (string, error)
 
diff --git a/pkg/retry/azure_error.go b/pkg/retry/azure_error.go
index d5ddae531..c7e9a71d1 100644
--- a/pkg/retry/azure_error.go
+++ b/pkg/retry/azure_error.go
@@ -425,3 +425,16 @@ func getOperationNotAllowedReason(msg string) string {
 	}
 	return OperationNotAllowed
 }
+
+// PartialUpdateError implements error interface. It is meant to be returned for errors with http status code of 2xx
+type PartialUpdateError struct {
+	message string
+}
+
+func NewPartialUpdateError(msg string) *PartialUpdateError {
+	return &PartialUpdateError{message: msg}
+}
+
+func (e *PartialUpdateError) Error() string {
+	return e.message
+}


Test patch:
diff --git a/pkg/provider/azure_controller_common_test.go b/pkg/provider/azure_controller_common_test.go
index 6ce2110c3..2474360cd 100644
--- a/pkg/provider/azure_controller_common_test.go
+++ b/pkg/provider/azure_controller_common_test.go
@@ -18,14 +18,19 @@ package provider
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"reflect"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-03-01/compute"
 	"github.com/Azure/go-autorest/autorest/azure"
+	autorestmocks "github.com/Azure/go-autorest/autorest/mocks"
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -44,6 +49,41 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
+var (
+	operationPreemptedError      = retry.NewError(false, errors.New(`Code="OperationPreempted" Message="Operation execution has been preempted by a more recent operation."`))
+	conflictingUserInputError    = retry.NewError(false, errors.New(`Code="ConflictingUserInput" Message="Cannot attach the disk pvc-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx to VM /subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx/resourceGroups/test-rg/providers/Microsoft.Compute/virtualMachineScaleSets/aks-nodepool0-00000000-vmss/virtualMachines/aks-nodepool0-00000000-vmss_0 because it is already attached to VM /subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx/resourceGroups/test-rg/providers/Microsoft.Compute/virtualMachineScaleSets/aks-nodepool0-00000000-vmss/virtualMachines/aks-nodepool0-00000000-vmss_1. A disk can be attached to only one VM at a time."`))
+	vmExtensionProvisioningError = retry.NewError(false, errors.New(`Code="VMExtensionProvisioningError" Message="Multiple VM extensions failed to be provisioned on the VM. Please see the VM extension instance view for other failures. The first extension failed due to the error: VM 'aks-nodepool0-00000000-vmss_00' has not reported status for VM agent or extensions. Verify that the OS is up and healthy, the VM has a running VM agent, and that it can establish outbound connections to Azure storage. Please refer to https://aka.ms/vmextensionlinuxtroubleshoot for additional VM agent troubleshooting information."`))
+)
+
+func fakeUpdateAsync(statusCode int) func(context.Context, string, string, compute.VirtualMachineUpdate, string) (*azure.Future, *retry.Error) {
+	return func(ctx context.Context, resourceGroup, nodeName string, parameters compute.VirtualMachineUpdate, source string) (*azure.Future, *retry.Error) {
+		vm := &compute.VirtualMachine{
+			Name:                     &nodeName,
+			Plan:                     parameters.Plan,
+			VirtualMachineProperties: parameters.VirtualMachineProperties,
+			Identity:                 parameters.Identity,
+			Zones:                    parameters.Zones,
+			Tags:                     parameters.Tags,
+		}
+		s, err := json.Marshal(vm)
+		if err != nil {
+			return nil, retry.NewError(false, err)
+		}
+
+		body := autorestmocks.NewBodyWithBytes(s)
+
+		r := autorestmocks.NewResponseWithBodyAndStatus(body, statusCode, strconv.Itoa(statusCode))
+		r.Request.Method = http.MethodPut
+
+		f, err := azure.NewFutureFromResponse(r)
+		if err != nil {
+			return nil, retry.NewError(false, err)
+		}
+
+		return &f, nil
+	}
+}
+
 func TestCommonAttachDisk(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -51,22 +91,46 @@ func TestCommonAttachDisk(t *testing.T) {
 	ctx, cancel := getContextWithCancel()
 	defer cancel()
 
+	initVM := func(testCloud *Cloud, expectedVMs []compute.VirtualMachine) {
+		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		for _, vm := range expectedVMs {
+			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, *vm.Name, gomock.Any()).Return(vm, nil).AnyTimes()
+		}
+		if len(expectedVMs) == 0 {
+			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any()).Return(compute.VirtualMachine{}, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
+		}
+	}
+
+	defaultSetup := func(testCloud *Cloud, expectedVMs []compute.VirtualMachine, statusCode int, result *retry.Error) {
+		initVM(testCloud, expectedVMs)
+		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+		mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(statusCode)).MaxTimes(1)
+		mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).MaxTimes(1)
+		mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result).MaxTimes(1)
+	}
+
 	maxShare := int32(1)
 	goodInstanceID := fmt.Sprintf("/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/%s", "vm1")
 	diskEncryptionSetID := fmt.Sprintf("/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Compute/diskEncryptionSets/%s", "diskEncryptionSet-name")
 	testTags := make(map[string]*string)
 	testTags[WriteAcceleratorEnabled] = to.StringPtr("true")
 	testCases := []struct {
-		desc            string
-		diskName        string
-		existedDisk     *compute.Disk
-		nodeName        types.NodeName
-		vmList          map[string]string
-		isDataDisksFull bool
-		isBadDiskURI    bool
-		isDiskUsed      bool
-		expectedErr     bool
-		expectedLun     int32
+		desc                 string
+		diskName             string
+		existedDisk          *compute.Disk
+		nodeName             types.NodeName
+		vmList               map[string]string
+		isDataDisksFull      bool
+		isBadDiskURI         bool
+		isDiskUsed           bool
+		setup                func(testCloud *Cloud, expectedVMs []compute.VirtualMachine, statusCode int, result *retry.Error)
+		expectErr            bool
+		isAcceptedErr        bool
+		isContextDeadlineErr bool
+		statusCode           int
+		waitResult           *retry.Error
+		expectedLun          int32
+		contextDuration      time.Duration
 	}{
 		{
 			desc:        "correct LUN and no error shall be returned if disk is nil",
@@ -75,7 +139,8 @@ func TestCommonAttachDisk(t *testing.T) {
 			diskName:    "disk-name",
 			existedDisk: nil,
 			expectedLun: 3,
-			expectedErr: false,
+			expectErr:   false,
+			statusCode:  200,
 		},
 		{
 			desc:        "LUN -1 and error shall be returned if there's no such instance corresponding to given nodeName",
@@ -83,7 +148,7 @@ func TestCommonAttachDisk(t *testing.T) {
 			diskName:    "disk-name",
 			existedDisk: &compute.Disk{Name: to.StringPtr("disk-name")},
 			expectedLun: -1,
-			expectedErr: true,
+			expectErr:   true,
 		},
 		{
 			desc:            "LUN -1 and error shall be returned if there's no available LUN for instance",
@@ -93,7 +158,7 @@ func TestCommonAttachDisk(t *testing.T) {
 			diskName:        "disk-name",
 			existedDisk:     &compute.Disk{Name: to.StringPtr("disk-name")},
 			expectedLun:     -1,
-			expectedErr:     true,
+			expectErr:       true,
 		},
 		{
 			desc:     "correct LUN and no error shall be returned if everything is good",
@@ -108,7 +173,8 @@ func TestCommonAttachDisk(t *testing.T) {
 				},
 				Tags: testTags},
 			expectedLun: 3,
-			expectedErr: false,
+			expectErr:   false,
+			statusCode:  200,
 		},
 		{
 			desc:     "an error shall be returned if disk state is not Unattached",
@@ -123,7 +189,7 @@ func TestCommonAttachDisk(t *testing.T) {
 				},
 				Tags: testTags},
 			expectedLun: -1,
-			expectedErr: true,
+			expectErr:   true,
 		},
 		{
 			desc:        "an error shall be returned if attach an already attached disk with good ManagedBy instance id",
@@ -132,55 +198,233 @@ func TestCommonAttachDisk(t *testing.T) {
 			diskName:    "disk-name",
 			existedDisk: &compute.Disk{Name: to.StringPtr("disk-name"), ManagedBy: to.StringPtr(goodInstanceID), DiskProperties: &compute.DiskProperties{MaxShares: &maxShare}},
 			expectedLun: -1,
-			expectedErr: true,
+			expectErr:   true,
+		},
+		{
+			desc:          "should return a PartialUpdateError type when storage configuration was accepted but wait fails with an error",
+			vmList:        map[string]string{"vm1": "PowerState/Running"},
+			nodeName:      "vm1",
+			diskName:      "disk-name",
+			existedDisk:   nil,
+			expectedLun:   -1,
+			expectErr:     true,
+			statusCode:    200,
+			waitResult:    conflictingUserInputError,
+			isAcceptedErr: true,
 		},
 		{
-			desc:        "an error shall be returned if attach an already attached disk with bad ManagedBy instance id",
+			desc:        "should not return a PartialUpdateError type when storage configuration was not accepted",
 			vmList:      map[string]string{"vm1": "PowerState/Running"},
 			nodeName:    "vm1",
 			diskName:    "disk-name",
-			existedDisk: &compute.Disk{Name: to.StringPtr("disk-name"), ManagedBy: to.StringPtr("test"), DiskProperties: &compute.DiskProperties{MaxShares: &maxShare}},
+			existedDisk: nil,
 			expectedLun: -1,
-			expectedErr: true,
+			expectErr:   true,
+			statusCode:  400,
+			waitResult:  conflictingUserInputError,
+		},
+		{
+			desc:        "should retry on OperationPreempted error and succeed",
+			vmList:      map[string]string{"vm1": "PowerState/Running"},
+			nodeName:    "vm1",
+			diskName:    "disk-name",
+			existedDisk: nil,
+			expectedLun: 3,
+			expectErr:   false,
+			statusCode:  200,
+			waitResult:  operationPreemptedError,
+			setup: func(testCloud *Cloud, expectedVMs []compute.VirtualMachine, statusCode int, result *retry.Error) {
+				defaultSetup(testCloud, expectedVMs, statusCode, result)
+				mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+				mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(200)).AnyTimes()
+				gomock.InOrder(
+					mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result),
+					mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(nil),
+				)
+			},
+		},
+		{
+			desc:        "should retry on OperationPreempted error until context deadline and return context.DeadlineExceeded error",
+			vmList:      map[string]string{"vm1": "PowerState/Running"},
+			nodeName:    "vm1",
+			diskName:    "disk-name",
+			existedDisk: nil,
+			expectedLun: -1,
+			expectErr:   true,
+			statusCode:  200,
+			waitResult:  operationPreemptedError,
+			setup: func(testCloud *Cloud, expectedVMs []compute.VirtualMachine, statusCode int, result *retry.Error) {
+				defaultSetup(testCloud, expectedVMs, statusCode, result)
+				mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+				mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(200)).AnyTimes()
+				mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result).AnyTimes()
+			},
+			contextDuration:      time.Second,
+			isContextDeadlineErr: true,
 		},
 	}
 
 	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(),
-			diskOpRateLimiter:     flowcontrol.NewTokenBucketRateLimiter(10, 20),
-		}
-		diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
-			testCloud.SubscriptionID, testCloud.ResourceGroup, test.diskName)
-		if test.isBadDiskURI {
-			diskURI = fmt.Sprintf("/baduri/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
+		tt := test
+		t.Run(tt.desc, func(t *testing.T) {
+			testCloud := GetTestCloud(ctrl)
+			diskURI := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
 				testCloud.SubscriptionID, testCloud.ResourceGroup, test.diskName)
-		}
-		expectedVMs := setTestVirtualMachines(testCloud, test.vmList, test.isDataDisksFull)
-		if test.isDiskUsed {
-			vm0 := setTestVirtualMachines(testCloud, map[string]string{"vm0": "PowerState/Running"}, test.isDataDisksFull)[0]
-			expectedVMs = append(expectedVMs, vm0)
-		}
+			if tt.isBadDiskURI {
+				diskURI = fmt.Sprintf("/baduri/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/disks/%s",
+					testCloud.SubscriptionID, testCloud.ResourceGroup, tt.diskName)
+			}
+			expectedVMs := setTestVirtualMachines(testCloud, tt.vmList, tt.isDataDisksFull)
+			if tt.isDiskUsed {
+				vm0 := setTestVirtualMachines(testCloud, map[string]string{"vm0": "PowerState/Running"}, tt.isDataDisksFull)[0]
+				expectedVMs = append(expectedVMs, vm0)
+			}
+			if tt.setup == nil {
+				defaultSetup(testCloud, expectedVMs, tt.statusCode, tt.waitResult)
+			} else {
+				tt.setup(testCloud, expectedVMs, tt.statusCode, tt.waitResult)
+			}
+
+			if tt.contextDuration > 0 {
+				oldCtx := ctx
+				ctx, cancel = context.WithTimeout(oldCtx, tt.contextDuration)
+				defer cancel()
+				defer func() {
+					ctx = oldCtx
+				}()
+			}
+
+			lun, err := testCloud.AttachDisk(ctx, true, "", diskURI, tt.nodeName, compute.CachingTypesReadOnly, tt.existedDisk)
+
+			assert.Equal(t, tt.expectedLun, lun, "TestCase[%d]: %s", i, tt.desc)
+			assert.Equal(t, tt.expectErr, err != nil, "TestCase[%d]: %s, return error: %v", i, tt.desc, err)
+
+			if tt.isAcceptedErr {
+				assert.IsType(t, &retry.PartialUpdateError{}, err)
+			} else {
+				var partialUpdateError *retry.PartialUpdateError
+				ok := errors.As(err, &partialUpdateError)
+				assert.False(t, ok, "the returned error should not be AcceptedError type")
+			}
+
+			assert.Equal(t, tt.isContextDeadlineErr, errors.Is(err, context.DeadlineExceeded))
+		})
+	}
+}
+
+func TestWaitForUpdateResult(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	defaultSetup := func(testCloud *Cloud, statusCode int, result *retry.Error) {
 		mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
-		for _, vm := range expectedVMs {
-			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, *vm.Name, gomock.Any()).Return(vm, nil).AnyTimes()
-		}
-		if len(expectedVMs) == 0 {
-			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any()).Return(compute.VirtualMachine{}, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
-		}
-		mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
-		mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(&azure.Future{}, nil).AnyTimes()
-		mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.ResourceGroup, gomock.Any()).Return(nil).AnyTimes()
+		mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(statusCode)).MaxTimes(1)
+		mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).MaxTimes(1)
+		mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result).MaxTimes(1)
+	}
 
-		lun, err := common.AttachDisk(ctx, true, "", diskURI, test.nodeName, compute.CachingTypesReadOnly, test.existedDisk)
-		assert.Equal(t, test.expectedLun, lun, "TestCase[%d]: %s", i, test.desc)
-		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s, return error: %v", i, test.desc, err)
+	testCases := []struct {
+		desc              string
+		nodeName          string
+		statusCode        int
+		waitResult        *retry.Error
+		expectErr         bool
+		expectedErrorType error
+		setup             func(testCloud *Cloud, statusCode int, result *retry.Error)
+		contextDuration   time.Duration
+	}{
+		{
+			desc:              "should return a PartialUpdateError type when storage configuration was accepted but wait fails with an error",
+			nodeName:          "vm1",
+			expectErr:         true,
+			statusCode:        200,
+			waitResult:        conflictingUserInputError,
+			expectedErrorType: &retry.PartialUpdateError{},
+		},
+		{
+			desc:              "should return a PartialUpdateError type when storage configuration was accepted but wait fails with an error",
+			nodeName:          "vm1",
+			expectErr:         true,
+			statusCode:        200,
+			waitResult:        conflictingUserInputError,
+			expectedErrorType: &retry.PartialUpdateError{},
+		},
+		{
+			desc:              "should not return a PartialUpdateError type when storage configuration was not accepted",
+			nodeName:          "vm1",
+			expectErr:         true,
+			statusCode:        400,
+			waitResult:        conflictingUserInputError,
+			expectedErrorType: conflictingUserInputError.Error(),
+		},
+		{
+			desc:       "should retry on OperationPreempted error and succeed",
+			nodeName:   "vm1",
+			expectErr:  false,
+			statusCode: 200,
+			waitResult: operationPreemptedError,
+			setup: func(testCloud *Cloud, statusCode int, result *retry.Error) {
+				defaultSetup(testCloud, statusCode, result)
+				mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+				mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(200))
+				gomock.InOrder(
+					mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result),
+					mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(nil),
+				)
+			},
+		},
+		{
+			desc:     "should retry on OperationPreempted error until context deadline and return context.DeadlineExceeded error",
+			nodeName: "vm1",
+
+			expectErr:  true,
+			statusCode: 200,
+			waitResult: operationPreemptedError,
+			setup: func(testCloud *Cloud, statusCode int, result *retry.Error) {
+				defaultSetup(testCloud, statusCode, result)
+				mockVMsClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
+				mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(fakeUpdateAsync(200)).AnyTimes()
+				mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), gomock.Any(), testCloud.resourceGroup, gomock.Any()).Return(result).AnyTimes()
+			},
+			contextDuration:   time.Second,
+			expectedErrorType: context.DeadlineExceeded,
+		},
+	}
+
+	for i, test := range testCases {
+		tt := test
+		t.Run(tt.desc, func(t *testing.T) {
+			testCloud := GetTestCloud(ctrl)
+			if tt.setup == nil {
+				defaultSetup(testCloud, tt.statusCode, tt.waitResult)
+			} else {
+				tt.setup(testCloud, tt.statusCode, tt.waitResult)
+			}
+
+			if tt.contextDuration > 0 {
+				oldCtx := ctx
+				ctx, cancel = context.WithTimeout(oldCtx, tt.contextDuration)
+				defer cancel()
+				defer func() {
+					ctx = oldCtx
+				}()
+			}
+
+			r := autorestmocks.NewResponseWithStatus(strconv.Itoa(tt.statusCode), tt.statusCode)
+			r.Request.Method = http.MethodPut
+			future, _ := azure.NewFutureFromResponse(r)
+
+			err := testCloud.waitForUpdateResult(ctx, testCloud.VMSet, types.NodeName(tt.nodeName), &future, nil)
+
+			assert.Equal(t, tt.expectErr, err != nil, "TestCase[%d]: %s, return error: %v", i, tt.desc, err)
+
+			if tt.expectErr {
+				assert.IsType(t, tt.expectedErrorType, err)
+			}
+		})
 	}
 }
 
@@ -406,15 +650,22 @@ func TestCommonUpdateVM(t *testing.T) {
 		if len(expectedVMs) == 0 {
 			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any()).Return(compute.VirtualMachine{}, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
 		}
+		r := autorestmocks.NewResponseWithStatus("200", 200)
+		r.Request.Method = http.MethodPut
+
+		future, err := azure.NewFutureFromResponse(r)
+
+		mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(&future, err).AnyTimes()
+
 		if test.isErrorRetriable {
 			testCloud.CloudProviderBackoff = true
 			testCloud.ResourceRequestBackoff = wait.Backoff{Steps: 1}
-			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(&retry.Error{HTTPStatusCode: http.StatusBadRequest, Retriable: true, RawError: fmt.Errorf("Retriable: true")}).AnyTimes()
+			mockVMsClient.EXPECT().WaitForUpdateResult(ctx, &future, testCloud.ResourceGroup, gomock.Any()).Return(&retry.Error{HTTPStatusCode: http.StatusBadRequest, Retriable: true, RawError: fmt.Errorf("Retriable: true")}).AnyTimes()
 		} else {
-			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			mockVMsClient.EXPECT().WaitForUpdateResult(ctx, &future, testCloud.ResourceGroup, gomock.Any()).Return(nil).AnyTimes()
 		}
 
-		err := common.UpdateVM(ctx, test.nodeName)
+		err = common.UpdateVM(ctx, test.nodeName)
 		assert.Equal(t, test.expectedErr, err != nil, "TestCase[%d]: %s, err: %v", i, test.desc, err)
 	}
 }
@@ -1086,3 +1337,123 @@ func TestDetachDiskRequestFuncs(t *testing.T) {
 		}
 	}
 }
+
+func TestGetAzureErrorCode(t *testing.T) {
+	testCases := []struct {
+		desc           string
+		err            error
+		expectedResult string
+	}{
+		{
+			desc:           "should return OperationPreempted",
+			err:            operationPreemptedError.Error(),
+			expectedResult: consts.OperationPreemptedErrorCode,
+		},
+		{
+			desc:           "should return VMExtensionprovisioning",
+			err:            vmExtensionProvisioningError.Error(),
+			expectedResult: "VMExtensionProvisioningError",
+		},
+		{
+			desc:           "should return ConflictingUserInput",
+			err:            conflictingUserInputError.Error(),
+			expectedResult: "ConflictingUserInput",
+		},
+	}
+	for i, test := range testCases {
+		tt := test
+		t.Run(tt.desc, func(t *testing.T) {
+			result := getAzureErrorCode(test.err)
+			assert.Equal(t, test.expectedResult, result, "TestCase[%d]", i, result)
+		})
+	}
+}
+
+func TestVmUpdateRequired(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	r := autorestmocks.NewResponseWithStatus("200", 200)
+	r.Request.Method = http.MethodPut
+
+	acceptedFuture, _ := azure.NewFutureFromResponse(r)
+
+	r = autorestmocks.NewResponseWithStatus("400", 400)
+	r.Request.Method = http.MethodPut
+
+	rejectedFuture, _ := azure.NewFutureFromResponse(r)
+
+	testCases := []struct {
+		desc           string
+		configAccepted bool
+		err            error
+		expectedResult bool
+	}{
+		{
+			desc:           "should return true if OperationPreemption error is returned with a http status code 2xx",
+			configAccepted: true,
+			err:            operationPreemptedError.Error(),
+			expectedResult: true,
+		},
+		{
+			desc:           "should return false if OperationPreemption error is returned with a http status code != 2xx",
+			configAccepted: false,
+			err:            operationPreemptedError.Error(),
+			expectedResult: false,
+		},
+		{
+			desc:           "should return false if ConflictingUserInputError error is returned even if http status code == 2xx",
+			configAccepted: true,
+			err:            conflictingUserInputError.Error(),
+			expectedResult: false,
+		},
+	}
+	for i, test := range testCases {
+		tt := test
+		t.Run(tt.desc, func(t *testing.T) {
+			var future *azure.Future
+			if tt.configAccepted {
+				future = &acceptedFuture
+			} else {
+				future = &rejectedFuture
+			}
+			result := vmUpdateRequired(future, test.err)
+			assert.Equalf(t, test.expectedResult, result, "TestCase[%d] returned %v", i, result)
+		})
+	}
+}
+
+func TestConfigAccepted(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc           string
+		statusCode     int
+		expectedResult bool
+	}{
+		{
+			desc:           "should return true if HTTP status code is 2xx",
+			statusCode:     200,
+			expectedResult: true,
+		},
+		{
+			desc:           "should return false if HTTP status code is not 2xx",
+			statusCode:     400,
+			expectedResult: false,
+		},
+	}
+
+	for i, testCase := range testCases {
+		tt := testCase
+		t.Run(tt.desc, func(t *testing.T) {
+			r := autorestmocks.NewResponseWithStatus(strconv.Itoa(tt.statusCode), tt.statusCode)
+			r.Request.Method = http.MethodPut
+
+			future, _ := azure.NewFutureFromResponse(r)
+
+			result := configAccepted(&future)
+			assert.Equalf(t, tt.expectedResult, result, "TestCase[%d] returned %v", i, result)
+		})
+	}
+}
diff --git a/pkg/provider/azure_controller_standard_test.go b/pkg/provider/azure_controller_standard_test.go
index 1c1c13a2a..3be4edba5 100644
--- a/pkg/provider/azure_controller_standard_test.go
+++ b/pkg/provider/azure_controller_standard_test.go
@@ -23,6 +23,8 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-03-01/compute"
+	"github.com/Azure/go-autorest/autorest/azure"
+	autorestmocks "github.com/Azure/go-autorest/autorest/mocks"
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -256,13 +258,21 @@ func TestStandardUpdateVM(t *testing.T) {
 			mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, *vm.Name, gomock.Any()).Return(vm, nil).AnyTimes()
 		}
 		mockVMsClient.EXPECT().Get(gomock.Any(), testCloud.ResourceGroup, "vm2", gomock.Any()).Return(compute.VirtualMachine{}, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
+
+		r := autorestmocks.NewResponseWithStatus("200", 200)
+		r.Request.Method = http.MethodPut
+
+		future, err := azure.NewFutureFromResponse(r)
+
+		mockVMsClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(&future, err).AnyTimes()
+
 		if test.isDetachFail {
-			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(&retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
+			mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), &future, testCloud.ResourceGroup, gomock.Any()).Return(&retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
 		} else {
-			mockVMsClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			mockVMsClient.EXPECT().WaitForUpdateResult(gomock.Any(), &future, testCloud.ResourceGroup, gomock.Any()).Return(nil).AnyTimes()
 		}
 
-		err := vmSet.UpdateVM(ctx, test.nodeName)
+		err = vmSet.UpdateVM(ctx, test.nodeName)
 		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 48b038c5c..17ef1931e 100644
--- a/pkg/provider/azure_controller_vmss_test.go
+++ b/pkg/provider/azure_controller_vmss_test.go
@@ -23,6 +23,8 @@ import (
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-03-01/compute"
+	"github.com/Azure/go-autorest/autorest/azure"
+	autorestmocks "github.com/Azure/go-autorest/autorest/mocks"
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -386,10 +388,18 @@ func TestUpdateVMWithVMSS(t *testing.T) {
 		}
 		mockVMSSVMClient := testCloud.VirtualMachineScaleSetVMsClient.(*mockvmssvmclient.MockInterface)
 		mockVMSSVMClient.EXPECT().List(gomock.Any(), testCloud.ResourceGroup, scaleSetName, gomock.Any()).Return(expectedVMSSVMs, nil).AnyTimes()
+
+		r := autorestmocks.NewResponseWithStatus("200", 200)
+		r.Request.Method = http.MethodPut
+
+		future, err := azure.NewFutureFromResponse(r)
+
+		mockVMSSVMClient.EXPECT().UpdateAsync(gomock.Any(), testCloud.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&future, err).AnyTimes()
+
 		if scaleSetName == strings.ToLower(string(fakeStatusNotFoundVMSSName)) {
-			mockVMSSVMClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, scaleSetName, gomock.Any(), gomock.Any(), gomock.Any()).Return(updatedVMSSVM, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
+			mockVMSSVMClient.EXPECT().WaitForUpdateResult(gomock.Any(), &future, testCloud.ResourceGroup, gomock.Any()).Return(updatedVMSSVM, &retry.Error{HTTPStatusCode: http.StatusNotFound, RawError: cloudprovider.InstanceNotFound}).AnyTimes()
 		} else {
-			mockVMSSVMClient.EXPECT().Update(gomock.Any(), testCloud.ResourceGroup, scaleSetName, gomock.Any(), gomock.Any(), gomock.Any()).Return(updatedVMSSVM, nil).AnyTimes()
+			mockVMSSVMClient.EXPECT().WaitForUpdateResult(gomock.Any(), &future, testCloud.ResourceGroup, gomock.Any()).Return(updatedVMSSVM, nil).AnyTimes()
 		}
 
 		mockVMClient := testCloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
diff --git a/pkg/provider/azure_controller_vmssflex_test.go b/pkg/provider/azure_controller_vmssflex_test.go
index ae1dc9e34..5eb06864a 100644
--- a/pkg/provider/azure_controller_vmssflex_test.go
+++ b/pkg/provider/azure_controller_vmssflex_test.go
@@ -22,6 +22,8 @@ import (
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-03-01/compute"
+	"github.com/Azure/go-autorest/autorest/azure"
+	autorestmocks "github.com/Azure/go-autorest/autorest/mocks"
 	"github.com/Azure/go-autorest/autorest/to"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -263,7 +265,14 @@ func TestUpdateVMWithVmssFlex(t *testing.T) {
 		mockVMClient := fs.VirtualMachinesClient.(*mockvmclient.MockInterface)
 		mockVMClient.EXPECT().ListVmssFlexVMsWithoutInstanceView(gomock.Any(), gomock.Any()).Return(tc.testVMListWithoutInstanceView, tc.vmListErr).AnyTimes()
 		mockVMClient.EXPECT().ListVmssFlexVMsWithOnlyInstanceView(gomock.Any(), gomock.Any()).Return(tc.testVMListWithOnlyInstanceView, tc.vmListErr).AnyTimes()
-		mockVMClient.EXPECT().Update(gomock.Any(), gomock.Any(), tc.vmName, gomock.Any(), "update_vm").Return(tc.vmssFlexVMUpdateError).AnyTimes()
+
+		r := autorestmocks.NewResponseWithStatus("200", 200)
+		r.Request.Method = http.MethodPut
+
+		future, err := azure.NewFutureFromResponse(r)
+
+		mockVMClient.EXPECT().UpdateAsync(gomock.Any(), gomock.Any(), tc.vmName, gomock.Any(), "update_vm").Return(&future, err).AnyTimes()
+		mockVMClient.EXPECT().WaitForUpdateResult(gomock.Any(), &future, gomock.Any(), gomock.Any()).Return(tc.vmssFlexVMUpdateError).AnyTimes()
 
 		err = fs.UpdateVM(ctx, tc.nodeName)