Repository: kubernetes-sigs/cloud-provider-azure
Pull request #555 (2021-03-13 13:11:32)

feat: implement cloud allocator for vmas
<!--  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?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
> /kind feature
> /kind flake
/kind feature

**What this PR does / why we need it**:
feat: implement cloud allocator for vmas. This would be useful after integrating with aks-engine because the availability set doesn't support manual scale.

**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`_*
-->
Ref: https://github.com/kubernetes-sigs/cloud-provider-azure/issues/462

**Special notes for your reviewer**:


**Release note**:
```
feat: implement cloud allocator for vmas
```


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/555#" title="Author self-approved">nilo19</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)~~ [nilo19]

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":[]} -->
k8s-ci-robot: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | 0a0d06d59a697a41378fe264ff8374ae92084ce5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-e2e-ccm/1370725306704334848) | `/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=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: /retest
k8s-ci-robot: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | 0a0d06d59a697a41378fe264ff8374ae92084ce5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-e2e-ccm/1371038975531159552) | `/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=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-unit | f65b5f6730d65ee8f2ae9874304a2c9a5abbb8c2 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-unit/1371385725966094336) | `/test pull-cloud-provider-azure-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-unit | f65b5f6730d65ee8f2ae9874304a2c9a5abbb8c2 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-unit/1371385725966094336) | `/test pull-cloud-provider-azure-unit`
pull-cloud-provider-azure-check | f65b5f6730d65ee8f2ae9874304a2c9a5abbb8c2 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-check/1371385725823488000) | `/test pull-cloud-provider-azure-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-unit | 31809cce009ba0aeca9b1bd2be6d73ca85b1920c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-unit/1371387598236291072) | `/test pull-cloud-provider-azure-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-unit | 31809cce009ba0aeca9b1bd2be6d73ca85b1920c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-unit/1371387598236291072) | `/test pull-cloud-provider-azure-unit`
pull-cloud-provider-azure-e2e-ccm | 31809cce009ba0aeca9b1bd2be6d73ca85b1920c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-e2e-ccm/1371387598198542336) | `/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=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | 9ef6987411bd5288060fdf72dd3953071c55a5c1 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-e2e-ccm/1371415831925231616) | `/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=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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: @nilo19: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm | 8254c5ab3bb6ed016f8692d9e88ab4833d6aaf19 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/555/pull-cloud-provider-azure-e2e-ccm/1371641192126615552) | `/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=555). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anilo19). 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/azureclients/vmasclient/azure_vmasclient.go b/pkg/azureclients/vmasclient/azure_vmasclient.go
new file mode 100644
index 000000000..f071ef615
--- /dev/null
+++ b/pkg/azureclients/vmasclient/azure_vmasclient.go
@@ -0,0 +1,306 @@
+/*
+Copyright 2021 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 vmasclient
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-12-01/compute"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/azure"
+	"github.com/Azure/go-autorest/autorest/to"
+
+	"k8s.io/client-go/util/flowcontrol"
+	"k8s.io/klog/v2"
+
+	azclients "sigs.k8s.io/cloud-provider-azure/pkg/azureclients"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/armclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+//var _ Interface = &Client{}
+
+// Client implements VMAS client Interface.
+type Client struct {
+	armClient      armclient.Interface
+	subscriptionID string
+	cloudName      string
+
+	// Rate limiting configures.
+	rateLimiterReader flowcontrol.RateLimiter
+	rateLimiterWriter flowcontrol.RateLimiter
+
+	// ARM throttling configures.
+	RetryAfterReader time.Time
+	RetryAfterWriter time.Time
+}
+
+// Get gets a AvailabilitySet.
+func (c *Client) Get(ctx context.Context, resourceGroupName string, vmasName string) (compute.AvailabilitySet, *retry.Error) {
+	mc := metrics.NewMetricContext("vmas", "get", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterReader.TryAccept() {
+		mc.RateLimitedCount()
+		return compute.AvailabilitySet{}, retry.GetRateLimitError(false, "VMASGet")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterReader.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("VMASGet", "client throttled", c.RetryAfterReader)
+		return compute.AvailabilitySet{}, rerr
+	}
+
+	result, rerr := c.getVMAS(ctx, resourceGroupName, vmasName)
+	_ = mc.Observe(rerr.Error())
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterReader = rerr.RetryAfter
+		}
+
+		return result, rerr
+	}
+
+	return result, nil
+}
+
+// getVMAS gets a AvailabilitySet.
+func (c *Client) getVMAS(ctx context.Context, resourceGroupName string, vmasName string) (compute.AvailabilitySet, *retry.Error) {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		"Microsoft.Compute/availabilitySets",
+		vmasName,
+	)
+	result := compute.AvailabilitySet{}
+
+	response, rerr := c.armClient.GetResource(ctx, resourceID, "")
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmas.get.request", resourceID, rerr.Error())
+		return result, rerr
+	}
+
+	err := autorest.Respond(
+		response,
+		azure.WithErrorUnlessStatusCode(http.StatusOK),
+		autorest.ByUnmarshallingJSON(&result))
+	if err != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmas.get.respond", resourceID, err)
+		return result, retry.GetError(response, err)
+	}
+
+	result.Response = autorest.Response{Response: response}
+	return result, nil
+}
+
+// New creates a new VMAS client with ratelimiting.
+func New(config *azclients.ClientConfig) *Client {
+	baseURI := config.ResourceManagerEndpoint
+	authorizer := config.Authorizer
+	apiVersion := APIVersion
+	if strings.EqualFold(config.CloudName, AzureStackCloudName) && !config.DisableAzureStackCloud {
+		apiVersion = AzureStackCloudAPIVersion
+	}
+	armClient := armclient.New(authorizer, baseURI, config.UserAgent, apiVersion, config.Location, config.Backoff)
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(config.RateLimitConfig)
+
+	klog.V(2).Infof("Azure AvailabilitySetsClient (read ops) using rate limit config: QPS=%g, bucket=%d",
+		config.RateLimitConfig.CloudProviderRateLimitQPS,
+		config.RateLimitConfig.CloudProviderRateLimitBucket)
+	klog.V(2).Infof("Azure AvailabilitySetsClient  (write ops) using rate limit config: QPS=%g, bucket=%d",
+		config.RateLimitConfig.CloudProviderRateLimitQPSWrite,
+		config.RateLimitConfig.CloudProviderRateLimitBucketWrite)
+
+	client := &Client{
+		armClient:         armClient,
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+		subscriptionID:    config.SubscriptionID,
+		cloudName:         config.CloudName,
+	}
+
+	return client
+}
+
+// List gets a list of AvailabilitySets in the resource group.
+func (c *Client) List(ctx context.Context, resourceGroupName string) ([]compute.AvailabilitySet, *retry.Error) {
+	mc := metrics.NewMetricContext("vmas", "list", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterReader.TryAccept() {
+		mc.RateLimitedCount()
+		return nil, retry.GetRateLimitError(false, "VMASList")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterReader.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("VMASList", "client throttled", c.RetryAfterReader)
+		return nil, rerr
+	}
+
+	result, rerr := c.listVMAS(ctx, resourceGroupName)
+	_ = mc.Observe(rerr.Error())
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterReader = rerr.RetryAfter
+		}
+
+		return result, rerr
+	}
+
+	return result, nil
+}
+
+// listVMAS gets a list of AvailabilitySets in the resource group.
+func (c *Client) listVMAS(ctx context.Context, resourceGroupName string) ([]compute.AvailabilitySet, *retry.Error) {
+	resourceID := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/availabilitySets",
+		autorest.Encode("path", c.subscriptionID),
+		autorest.Encode("path", resourceGroupName))
+	result := make([]compute.AvailabilitySet, 0)
+	page := &AvailabilitySetListResultPage{}
+	page.fn = c.listNextResults
+
+	resp, rerr := c.armClient.GetResource(ctx, resourceID, "")
+	defer c.armClient.CloseResponse(ctx, resp)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmas.list.request", resourceID, rerr.Error())
+		return result, rerr
+	}
+
+	var err error
+	page.vmaslr, err = c.listResponder(resp)
+	if err != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmas.list.respond", resourceID, err)
+		return result, retry.GetError(resp, err)
+	}
+
+	for {
+		result = append(result, page.Values()...)
+
+		// Abort the loop when there's no nextLink in the response.
+		if to.String(page.Response().NextLink) == "" {
+			break
+		}
+
+		if err = page.NextWithContext(ctx); err != nil {
+			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "vmas.list.next", resourceID, err)
+			return result, retry.GetError(page.Response().Response.Response, err)
+		}
+	}
+
+	return result, nil
+}
+
+func (c *Client) listResponder(resp *http.Response) (result compute.AvailabilitySetListResult, err error) {
+	err = autorest.Respond(
+		resp,
+		autorest.ByIgnoring(),
+		azure.WithErrorUnlessStatusCode(http.StatusOK),
+		autorest.ByUnmarshallingJSON(&result))
+	result.Response = autorest.Response{Response: resp}
+	return
+}
+
+// availabilitySetListResultPreparer prepares a request to retrieve the next set of results.
+// It returns nil if no more results exist.
+func (c *Client) availabilitySetListResultPreparer(ctx context.Context, vmaslr compute.AvailabilitySetListResult) (*http.Request, error) {
+	if vmaslr.NextLink == nil || len(to.String(vmaslr.NextLink)) < 1 {
+		return nil, nil
+	}
+
+	decorators := []autorest.PrepareDecorator{
+		autorest.WithBaseURL(to.String(vmaslr.NextLink)),
+	}
+	return c.armClient.PrepareGetRequest(ctx, decorators...)
+}
+
+// listNextResults retrieves the next set of results, if any.
+func (c *Client) listNextResults(ctx context.Context, lastResults compute.AvailabilitySetListResult) (result compute.AvailabilitySetListResult, err error) {
+	req, err := c.availabilitySetListResultPreparer(ctx, lastResults)
+	if err != nil {
+		return result, autorest.NewErrorWithError(err, "vmasclient", "listNextResults", nil, "Failure preparing next results request")
+	}
+	if req == nil {
+		return
+	}
+
+	resp, rerr := c.armClient.Send(ctx, req)
+	defer c.armClient.CloseResponse(ctx, resp)
+	if rerr != nil {
+		result.Response = autorest.Response{Response: resp}
+		return result, autorest.NewErrorWithError(rerr.Error(), "vmasclient", "listNextResults", resp, "Failure sending next results request")
+	}
+
+	result, err = c.listResponder(resp)
+	if err != nil {
+		err = autorest.NewErrorWithError(err, "vmasclient", "listNextResults", resp, "Failure responding to next results request")
+	}
+
+	return
+}
+
+// AvailabilitySetListResultPage  contains a page of AvailabilitySet values.
+type AvailabilitySetListResultPage struct {
+	fn     func(context.Context, compute.AvailabilitySetListResult) (compute.AvailabilitySetListResult, error)
+	vmaslr compute.AvailabilitySetListResult
+}
+
+// NextWithContext advances to the next page of values.  If there was an error making
+// the request the page does not advance and the error is returned.
+func (page *AvailabilitySetListResultPage) NextWithContext(ctx context.Context) (err error) {
+	next, err := page.fn(ctx, page.vmaslr)
+	if err != nil {
+		return err
+	}
+	page.vmaslr = next
+	return nil
+}
+
+// Next advances to the next page of values.  If there was an error making
+// the request the page does not advance and the error is returned.
+// Deprecated: Use NextWithContext() instead.
+func (page *AvailabilitySetListResultPage) Next() error {
+	return page.NextWithContext(context.Background())
+}
+
+// NotDone returns true if the page enumeration should be started or is not yet complete.
+func (page AvailabilitySetListResultPage) NotDone() bool {
+	return !page.vmaslr.IsEmpty()
+}
+
+// Response returns the raw server response from the last page request.
+func (page AvailabilitySetListResultPage) Response() compute.AvailabilitySetListResult {
+	return page.vmaslr
+}
+
+// Values returns the slice of values for the current page or nil if there are no values.
+func (page AvailabilitySetListResultPage) Values() []compute.AvailabilitySet {
+	if page.vmaslr.IsEmpty() {
+		return nil
+	}
+	return *page.vmaslr.Value
+}
diff --git a/pkg/azureclients/vmasclient/doc.go b/pkg/azureclients/vmasclient/doc.go
new file mode 100644
index 000000000..7f9dee7ed
--- /dev/null
+++ b/pkg/azureclients/vmasclient/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2021 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 vmasclient implements the client for VMAS.
+package vmasclient // import "sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient"
diff --git a/pkg/azureclients/vmasclient/interface.go b/pkg/azureclients/vmasclient/interface.go
new file mode 100644
index 000000000..5f9ba235e
--- /dev/null
+++ b/pkg/azureclients/vmasclient/interface.go
@@ -0,0 +1,44 @@
+/*
+Copyright 2021 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 vmasclient
+
+import (
+	"context"
+
+	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-12-01/compute"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+const (
+	// APIVersion is the API version for VMAS.
+	APIVersion = "2020-06-30"
+	// AzureStackCloudAPIVersion is the API version for Azure Stack
+	AzureStackCloudAPIVersion = "2019-07-01"
+	// AzureStackCloudName is the cloud name of Azure Stack
+	AzureStackCloudName = "AZURESTACKCLOUD"
+)
+
+// Interface is the client interface for AvailabilitySet.
+// Don't forget to run the following command to generate the mock client:
+// mockgen -source=$GOPATH/src/sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient/interface.go -package=mockvmasclient Interface > $GOPATH/src/sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient/mockvmasclient/interface.go
+type Interface interface {
+	// Get gets a VirtualMachineScaleSet.
+	Get(ctx context.Context, resourceGroupName string, VMScaleSetName string) (result compute.AvailabilitySet, rerr *retry.Error)
+
+	// List gets a list of VirtualMachineScaleSets in the resource group.
+	List(ctx context.Context, resourceGroupName string) (result []compute.AvailabilitySet, rerr *retry.Error)
+}
diff --git a/pkg/azureclients/vmasclient/mockvmasclient/doc.go b/pkg/azureclients/vmasclient/mockvmasclient/doc.go
new file mode 100644
index 000000000..838085902
--- /dev/null
+++ b/pkg/azureclients/vmasclient/mockvmasclient/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2021 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 mockvmasclient implements the mock client for VMAS.
+package mockvmasclient // import "sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient/mockvmssclient"
diff --git a/pkg/azureclients/vmasclient/mockvmasclient/interface.go b/pkg/azureclients/vmasclient/mockvmasclient/interface.go
new file mode 100644
index 000000000..2e13e1e53
--- /dev/null
+++ b/pkg/azureclients/vmasclient/mockvmasclient/interface.go
@@ -0,0 +1,80 @@
+/*
+Copyright 2021 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 mockvmasclient
+
+import (
+	context "context"
+	reflect "reflect"
+
+	compute "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-12-01/compute"
+	gomock "github.com/golang/mock/gomock"
+
+	retry "sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+// MockInterface is a mock of Interface interface
+type MockInterface struct {
+	ctrl     *gomock.Controller
+	recorder *MockInterfaceMockRecorder
+}
+
+// MockInterfaceMockRecorder is the mock recorder for MockInterface
+type MockInterfaceMockRecorder struct {
+	mock *MockInterface
+}
+
+// NewMockInterface creates a new mock instance
+func NewMockInterface(ctrl *gomock.Controller) *MockInterface {
+	mock := &MockInterface{ctrl: ctrl}
+	mock.recorder = &MockInterfaceMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockInterface) EXPECT() *MockInterfaceMockRecorder {
+	return m.recorder
+}
+
+// Get mocks base method
+func (m *MockInterface) Get(ctx context.Context, resourceGroupName, VMScaleSetName string) (compute.AvailabilitySet, *retry.Error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Get", ctx, resourceGroupName, VMScaleSetName)
+	ret0, _ := ret[0].(compute.AvailabilitySet)
+	ret1, _ := ret[1].(*retry.Error)
+	return ret0, ret1
+}
+
+// Get indicates an expected call of Get
+func (mr *MockInterfaceMockRecorder) Get(ctx, resourceGroupName, VMScaleSetName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockInterface)(nil).Get), ctx, resourceGroupName, VMScaleSetName)
+}
+
+// List mocks base method
+func (m *MockInterface) List(ctx context.Context, resourceGroupName string) ([]compute.AvailabilitySet, *retry.Error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "List", ctx, resourceGroupName)
+	ret0, _ := ret[0].([]compute.AvailabilitySet)
+	ret1, _ := ret[1].(*retry.Error)
+	return ret0, ret1
+}
+
+// List indicates an expected call of List
+func (mr *MockInterfaceMockRecorder) List(ctx, resourceGroupName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockInterface)(nil).List), ctx, resourceGroupName)
+}
diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go
index ea54fd297..bf05d1a5f 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -115,11 +115,26 @@ const (
 	VMSetCIDRIPV4TagKey = "kubernetesNodeCIDRMaskIPV4"
 	// VMSetCIDRIPV6TagKey specifies the node ipv6 CIDR mask of the instances on the VMSS or VMAS
 	VMSetCIDRIPV6TagKey = "kubernetesNodeCIDRMaskIPV6"
+)
 
-	// TagsDelimiter is the delimiter of tags
-	TagsDelimiter = ","
-	// TagKeyValueDelimiter is the delimiter between keys and values in tagas
-	TagKeyValueDelimiter = "="
+// cache
+const (
+	// VMSSNameSeparator is the separator of the vmss names
+	VMSSNameSeparator = "_"
+	// VMSSKey is the key when querying vmss cache
+	VMSSKey = "k8svmssKey"
+	// VMASKey is the key when querying vmss cache
+	VMASKey = "k8svmasKey"
+	// AvailabilitySetNodesKey is the availability set nodes key
+	AvailabilitySetNodesKey = "k8sAvailabilitySetNodesKey"
+	// AvailabilitySetNodesCacheTTLDefaultInSeconds is the TTL of the availabilitySet node cache
+	AvailabilitySetNodesCacheTTLDefaultInSeconds = 900
+	// VMSSCacheTTLDefaultInSeconds is the TTL of the vmss cache
+	VMSSCacheTTLDefaultInSeconds = 600
+	// VMSSVirtualMachinesCacheTTLDefaultInSeconds is the TTL of the vmss vm cache
+	VMSSVirtualMachinesCacheTTLDefaultInSeconds = 600
+	// VMASCacheTTLDefaultInSeconds is the TTL of the vmas cache
+	VMASCacheTTLDefaultInSeconds = 600
 )
 
 // azure cloud config
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index 34289ae08..aed7d5b4e 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -56,6 +56,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/snapshotclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/storageaccountclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/subnetclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmsizeclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmssclient"
@@ -138,10 +139,6 @@ type Config struct {
 	// includes load balancer, security group and route table. The supported format is `a=b,c=d,...`. After updated
 	// this config, the old tags would be replaced by the new ones.
 	Tags string `json:"tags,omitempty" yaml:"tags,omitempty"`
-	// SystemTags determines the tag keys managed by cloud provider. If it is not set, no tags would be deleted if
-	// the `Tags` is changed. However, the old tags would be deleted if they are neither included in `Tags` nor
-	// in `SystemTags` after the update of `Tags`.
-	SystemTags string `json:"systemTags,omitempty" yaml:"systemTags,omitempty"`
 	// Sku of Load Balancer and Public IP. Candidate values are: basic and standard.
 	// If not set, it will be default to basic.
 	LoadBalancerSku string `json:"loadBalancerSku,omitempty" yaml:"loadBalancerSku,omitempty"`
@@ -209,6 +206,8 @@ type Config struct {
 	NsgCacheTTLInSeconds int `json:"nsgCacheTTLInSeconds,omitempty" yaml:"nsgCacheTTLInSeconds,omitempty"`
 	// RouteTableCacheTTLInSeconds sets the cache TTL for route table
 	RouteTableCacheTTLInSeconds int `json:"routeTableCacheTTLInSeconds,omitempty" yaml:"routeTableCacheTTLInSeconds,omitempty"`
+	// AvailabilitySetsCacheTTLInSeconds sets the cache TTL for VMAS
+	AvailabilitySetsCacheTTLInSeconds int `json:"availabilitySetsCacheTTLInSeconds,omitempty" yaml:"availabilitySetsCacheTTLInSeconds,omitempty"`
 	// RouteUpdateWaitingInSeconds is the delay time for waiting route updates to take effect. This waiting delay is added
 	// because the routes are not taken effect when the async route updating operation returns success. Default is 30 seconds.
 	RouteUpdateWaitingInSeconds int `json:"routeUpdateWaitingInSeconds,omitempty" yaml:"routeUpdateWaitingInSeconds,omitempty"`
@@ -248,6 +247,7 @@ type Cloud struct {
 	VirtualMachineScaleSetsClient   vmssclient.Interface
 	VirtualMachineScaleSetVMsClient vmssvmclient.Interface
 	VirtualMachineSizesClient       vmsizeclient.Interface
+	AvailabilitySetsClient          vmasclient.Interface
 
 	ResourceRequestBackoff wait.Backoff
 	metadata               *InstanceMetadataService
@@ -450,7 +450,10 @@ func (az *Cloud) InitializeCloudFromConfig(config *Config, fromSecret bool) erro
 			return err
 		}
 	} else {
-		az.VMSet = newAvailabilitySet(az)
+		az.VMSet, err = newAvailabilitySet(az)
+		if err != nil {
+			return err
+		}
 	}
 
 	az.vmCache, err = az.newVMCache()
@@ -588,13 +591,15 @@ func (az *Cloud) configAzureClients(
 	publicIPClientConfig := azClientConfig.WithRateLimiter(az.Config.PublicIPAddressRateLimit)
 	// TODO(ZeroMagic): add azurefileRateLimit
 	fileClientConfig := azClientConfig.WithRateLimiter(nil)
+	vmasClientConfig := azClientConfig.WithRateLimiter(az.Config.AvailabilitySetRateLimit)
 
-	// If uses network resources in different AAD Tenant, update Authorizer for VM/VMSS client config
+	// If uses network resources in different AAD Tenant, update Authorizer for VM/VMSS/VMAS client config
 	if multiTenantServicePrincipalToken != nil {
 		multiTenantServicePrincipalTokenAuthorizer := autorest.NewMultiTenantServicePrincipalTokenAuthorizer(multiTenantServicePrincipalToken)
 		vmClientConfig.Authorizer = multiTenantServicePrincipalTokenAuthorizer
 		vmssClientConfig.Authorizer = multiTenantServicePrincipalTokenAuthorizer
 		vmssVMClientConfig.Authorizer = multiTenantServicePrincipalTokenAuthorizer
+		vmasClientConfig.Authorizer = multiTenantServicePrincipalTokenAuthorizer
 	}
 
 	// If uses network resources in different AAD Tenant, update SubscriptionID and Authorizer for network resources client config
@@ -631,6 +636,7 @@ func (az *Cloud) configAzureClients(
 	az.SecurityGroupsClient = securitygroupclient.New(securityGroupClientConfig)
 	az.PublicIPAddressesClient = publicipclient.New(publicIPClientConfig)
 	az.FileClient = fileclient.New(fileClientConfig)
+	az.AvailabilitySetsClient = vmasclient.New(vmasClientConfig)
 }
 
 func (az *Cloud) getAzureClientConfig(servicePrincipalToken *adal.ServicePrincipalToken) *azclients.ClientConfig {
diff --git a/pkg/provider/azure_controller_common.go b/pkg/provider/azure_controller_common.go
index baee55b69..9e0dc7b62 100644
--- a/pkg/provider/azure_controller_common.go
+++ b/pkg/provider/azure_controller_common.go
@@ -186,10 +186,6 @@ func (c *controllerCommon) AttachDisk(isManagedDisk bool, diskName, diskURI stri
 				disk.DiskProperties.Encryption.DiskEncryptionSetID != nil {
 				diskEncryptionSetID = *disk.DiskProperties.Encryption.DiskEncryptionSetID
 			}
-
-			if disk.DiskProperties.DiskState != compute.Unattached && (disk.MaxShares == nil || *disk.MaxShares <= 1) {
-				return -1, fmt.Errorf("state of disk(%s) is %s, not in expected %s state", diskURI, disk.DiskProperties.DiskState, compute.Unattached)
-			}
 		}
 
 		if v, ok := disk.Tags[WriteAcceleratorEnabled]; ok {
diff --git a/pkg/provider/azure_fakes.go b/pkg/provider/azure_fakes.go
index 5b4955af4..6b116228e 100644
--- a/pkg/provider/azure_fakes.go
+++ b/pkg/provider/azure_fakes.go
@@ -98,7 +98,7 @@ func GetTestCloud(ctrl *gomock.Controller) (az *Cloud) {
 	az.VirtualMachineScaleSetsClient = mockvmssclient.NewMockInterface(ctrl)
 	az.VirtualMachineScaleSetVMsClient = mockvmssvmclient.NewMockInterface(ctrl)
 	az.VirtualMachinesClient = mockvmclient.NewMockInterface(ctrl)
-	az.VMSet = newAvailabilitySet(az)
+	az.VMSet, _ = newAvailabilitySet(az)
 	az.vmCache, _ = az.newVMCache()
 	az.lbCache, _ = az.newLBCache()
 	az.nsgCache, _ = az.newNSGCache()
diff --git a/pkg/provider/azure_loadbalancer.go b/pkg/provider/azure_loadbalancer.go
index df4d486a8..afccd1597 100644
--- a/pkg/provider/azure_loadbalancer.go
+++ b/pkg/provider/azure_loadbalancer.go
@@ -2264,8 +2264,9 @@ func shouldReleaseExistingOwnedPublicIP(existingPip *network.PublicIPAddress, lb
 		(ipTagRequest.IPTagsRequestedByAnnotation && !areIPTagsEquivalent(currentIPTags, ipTagRequest.IPTags))
 }
 
-// ensurePIPTagged ensures the public IP of the service is tagged as configured
+//  ensurePIPTagged ensures the public IP of the service is tagged as configured
 func (az *Cloud) ensurePIPTagged(service *v1.Service, pip *network.PublicIPAddress) bool {
+	changed := false
 	configTags := parseTags(az.Tags)
 	annotationTags := make(map[string]*string)
 	if _, ok := service.Annotations[consts.ServiceAnnotationAzurePIPTags]; ok {
@@ -2274,7 +2275,6 @@ func (az *Cloud) ensurePIPTagged(service *v1.Service, pip *network.PublicIPAddre
 	for k, v := range annotationTags {
 		configTags[k] = v
 	}
-
 	// include the cluster name and service names tags when comparing
 	var clusterName, serviceNames *string
 	if v, ok := pip.Tags[consts.ClusterNameKey]; ok {
@@ -2289,10 +2289,12 @@ func (az *Cloud) ensurePIPTagged(service *v1.Service, pip *network.PublicIPAddre
 	if serviceNames != nil {
 		configTags[consts.ServiceTagKey] = serviceNames
 	}
-
-	tags, changed := az.reconcileTags(pip.Tags, configTags)
-	pip.Tags = tags
-
+	for k, v := range configTags {
+		if vv, ok := pip.Tags[k]; !ok || !strings.EqualFold(to.String(v), to.String(vv)) {
+			pip.Tags[k] = v
+			changed = true
+		}
+	}
 	return changed
 }
 
@@ -2819,6 +2821,7 @@ func unbindServiceFromPIP(pip *network.PublicIPAddress, serviceName string) erro
 
 // ensureLoadBalancerTagged ensures every load balancer in the resource group is tagged as configured
 func (az *Cloud) ensureLoadBalancerTagged(lb *network.LoadBalancer) bool {
+	changed := false
 	if az.Tags == "" {
 		return false
 	}
@@ -2826,15 +2829,18 @@ func (az *Cloud) ensureLoadBalancerTagged(lb *network.LoadBalancer) bool {
 	if lb.Tags == nil {
 		lb.Tags = make(map[string]*string)
 	}
-
-	tags, changed := az.reconcileTags(lb.Tags, tags)
-	lb.Tags = tags
-
+	for k, v := range tags {
+		if vv, ok := lb.Tags[k]; !ok || !strings.EqualFold(to.String(v), to.String(vv)) {
+			lb.Tags[k] = v
+			changed = true
+		}
+	}
 	return changed
 }
 
 // ensureSecurityGroupTagged ensures the security group is tagged as configured
 func (az *Cloud) ensureSecurityGroupTagged(sg *network.SecurityGroup) bool {
+	changed := false
 	if az.Tags == "" {
 		return false
 	}
@@ -2842,9 +2848,11 @@ func (az *Cloud) ensureSecurityGroupTagged(sg *network.SecurityGroup) bool {
 	if sg.Tags == nil {
 		sg.Tags = make(map[string]*string)
 	}
-
-	tags, changed := az.reconcileTags(sg.Tags, tags)
-	sg.Tags = tags
-
+	for k, v := range tags {
+		if vv, ok := sg.Tags[k]; !ok || !strings.EqualFold(to.String(v), to.String(vv)) {
+			sg.Tags[k] = v
+			changed = true
+		}
+	}
 	return changed
 }
diff --git a/pkg/provider/azure_ratelimit.go b/pkg/provider/azure_ratelimit.go
index aa8b441a4..94e6acfb9 100644
--- a/pkg/provider/azure_ratelimit.go
+++ b/pkg/provider/azure_ratelimit.go
@@ -40,6 +40,7 @@ type CloudProviderRateLimitConfig struct {
 	SnapshotRateLimit               *azclients.RateLimitConfig `json:"snapshotRateLimit,omitempty" yaml:"snapshotRateLimit,omitempty"`
 	VirtualMachineScaleSetRateLimit *azclients.RateLimitConfig `json:"virtualMachineScaleSetRateLimit,omitempty" yaml:"virtualMachineScaleSetRateLimit,omitempty"`
 	VirtualMachineSizeRateLimit     *azclients.RateLimitConfig `json:"virtualMachineSizesRateLimit,omitempty" yaml:"virtualMachineSizesRateLimit,omitempty"`
+	AvailabilitySetRateLimit        *azclients.RateLimitConfig `json:"availabilitySetRateLimit,omitempty" yaml:"availabilitySetRateLimit,omitempty"`
 }
 
 // InitializeCloudProviderRateLimitConfig initializes rate limit configs.
@@ -76,6 +77,7 @@ func InitializeCloudProviderRateLimitConfig(config *CloudProviderRateLimitConfig
 	config.SnapshotRateLimit = overrideDefaultRateLimitConfig(&config.RateLimitConfig, config.SnapshotRateLimit)
 	config.VirtualMachineScaleSetRateLimit = overrideDefaultRateLimitConfig(&config.RateLimitConfig, config.VirtualMachineScaleSetRateLimit)
 	config.VirtualMachineSizeRateLimit = overrideDefaultRateLimitConfig(&config.RateLimitConfig, config.VirtualMachineSizeRateLimit)
+	config.AvailabilitySetRateLimit = overrideDefaultRateLimitConfig(&config.RateLimitConfig, config.AvailabilitySetRateLimit)
 }
 
 // overrideDefaultRateLimitConfig overrides the default CloudProviderRateLimitConfig.
diff --git a/pkg/provider/azure_routes.go b/pkg/provider/azure_routes.go
index e72ab7b57..0965d2c3c 100644
--- a/pkg/provider/azure_routes.go
+++ b/pkg/provider/azure_routes.go
@@ -505,18 +505,21 @@ func cidrtoRfc1035(cidr string) string {
 	return cidr
 }
 
-// ensureRouteTableTagged ensures the route table is tagged as configured
+//  ensureRouteTableTagged ensures the route table is tagged as configured
 func (az *Cloud) ensureRouteTableTagged(rt *network.RouteTable) (map[string]*string, bool) {
 	if az.Tags == "" {
 		return nil, false
 	}
+	changed := false
 	tags := parseTags(az.Tags)
 	if rt.Tags == nil {
 		rt.Tags = make(map[string]*string)
 	}
-
-	tags, changed := az.reconcileTags(rt.Tags, tags)
-	rt.Tags = tags
-
+	for k, v := range tags {
+		if vv, ok := rt.Tags[k]; !ok || !strings.EqualFold(to.String(v), to.String(vv)) {
+			rt.Tags[k] = v
+			changed = true
+		}
+	}
 	return rt.Tags, changed
 }
diff --git a/pkg/provider/azure_standard.go b/pkg/provider/azure_standard.go
index 53d0d2dd3..12c7a3f83 100644
--- a/pkg/provider/azure_standard.go
+++ b/pkg/provider/azure_standard.go
@@ -24,6 +24,8 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
+	"time"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 
@@ -409,13 +411,67 @@ func MakeCRC32(str string) string {
 // availabilitySet implements VMSet interface for Azure availability sets.
 type availabilitySet struct {
 	*Cloud
+
+	vmasCache *azcache.TimedCache
+}
+
+type availabilitySetEntry struct {
+	vmas          *compute.AvailabilitySet
+	resourceGroup string
+}
+
+func (as *availabilitySet) newVMASCache() (*azcache.TimedCache, error) {
+	getter := func(key string) (interface{}, error) {
+		localCache := &sync.Map{}
+
+		allResourceGroups, err := as.GetResourceGroups()
+		if err != nil {
+			return nil, err
+		}
+
+		for _, resourceGroup := range allResourceGroups.List() {
+			allAvailabilitySets, rerr := as.AvailabilitySetsClient.List(context.Background(), resourceGroup)
+			if rerr != nil {
+				klog.Errorf("AvailabilitySetsClient.List failed: %v", rerr)
+				return nil, rerr.Error()
+			}
+
+			for i := range allAvailabilitySets {
+				vmas := allAvailabilitySets[i]
+				if strings.EqualFold(to.String(vmas.Name), "") {
+					klog.Warning("failed to get the name of the VMAS")
+					continue
+				}
+				localCache.Store(to.String(vmas.Name), &availabilitySetEntry{
+					vmas:          &vmas,
+					resourceGroup: resourceGroup,
+				})
+			}
+		}
+
+		return localCache, nil
+	}
+
+	if as.Config.AvailabilitySetsCacheTTLInSeconds == 0 {
+		as.Config.AvailabilitySetsCacheTTLInSeconds = consts.VMASCacheTTLDefaultInSeconds
+	}
+
+	return azcache.NewTimedcache(time.Duration(as.Config.AvailabilitySetsCacheTTLInSeconds)*time.Second, getter)
 }
 
 // newStandardSet creates a new availabilitySet.
-func newAvailabilitySet(az *Cloud) VMSet {
-	return &availabilitySet{
+func newAvailabilitySet(az *Cloud) (VMSet, error) {
+	as := &availabilitySet{
 		Cloud: az,
 	}
+
+	var err error
+	as.vmasCache, err = as.newVMASCache()
+	if err != nil {
+		return nil, err
+	}
+
+	return as, nil
 }
 
 // GetInstanceIDByNodeName gets the cloud provider ID by node name.
@@ -1062,7 +1118,83 @@ func (as *availabilitySet) GetNodeNameByIPConfigurationID(ipConfigurationID stri
 	return vmName, strings.ToLower(asName), nil
 }
 
+func (as *availabilitySet) getAvailabilitySetByNodeName(nodeName string, crt azcache.AzureCacheReadType) (*compute.AvailabilitySet, error) {
+	cached, err := as.vmasCache.Get(consts.VMASKey, crt)
+	if err != nil {
+		return nil, err
+	}
+	vmasList := cached.(*sync.Map)
+
+	if vmasList == nil {
+		klog.Warning("Couldn't get all vmas from cache")
+		return nil, nil
+	}
+
+	var result *compute.AvailabilitySet
+	vmasList.Range(func(_, value interface{}) bool {
+		vmasEntry := value.(*availabilitySetEntry)
+		vmas := vmasEntry.vmas
+		if vmas != nil && vmas.AvailabilitySetProperties != nil && vmas.VirtualMachines != nil {
+			for _, vmIDRef := range *vmas.VirtualMachines {
+				if vmIDRef.ID != nil {
+					matches := vmIDRE.FindStringSubmatch(to.String(vmIDRef.ID))
+					if len(matches) != 2 {
+						err = fmt.Errorf("invalid vm ID %s", to.String(vmIDRef.ID))
+						return false
+					}
+
+					vmName := matches[1]
+					if strings.EqualFold(vmName, nodeName) {
+						result = vmas
+						return false
+					}
+				}
+			}
+		}
+
+		return true
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	if result == nil {
+		klog.Warningf("failed to find the vmas of node %s", nodeName)
+		return nil, cloudprovider.InstanceNotFound
+	}
+
+	return result, nil
+}
+
 // GetNodeCIDRMaskByProviderID returns the node CIDR subnet mask by provider ID.
 func (as *availabilitySet) GetNodeCIDRMasksByProviderID(providerID string) (int, int, error) {
-	return 0, 0, cloudprovider.NotImplemented
+	nodeName, err := as.GetNodeNameByProviderID(providerID)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	vmas, err := as.getAvailabilitySetByNodeName(string(nodeName), azcache.CacheReadTypeDefault)
+	if err != nil {
+		if errors.Is(err, cloudprovider.InstanceNotFound) {
+			return consts.DefaultNodeMaskCIDRIPv4, consts.DefaultNodeMaskCIDRIPv6, nil
+		}
+		return 0, 0, err
+	}
+
+	var ipv4Mask, ipv6Mask int
+	if v4, ok := vmas.Tags[consts.VMSetCIDRIPV4TagKey]; ok && v4 != nil {
+		ipv4Mask, err = strconv.Atoi(to.String(v4))
+		if err != nil {
+			klog.Errorf("GetNodeCIDRMasksByProviderID: error when paring the value of the ipv4 mask size %s: %v", to.String(v4), err)
+		}
+	}
+	if v6, ok := vmas.Tags[consts.VMSetCIDRIPV6TagKey]; ok && v6 != nil {
+		ipv6Mask, err = strconv.Atoi(to.String(v6))
+		if err != nil {
+			klog.Errorf("GetNodeCIDRMasksByProviderID: error when paring the value of the ipv6 mask size%s: %v", to.String(v6), err)
+		}
+	}
+
+	return ipv4Mask, ipv6Mask, nil
 }
diff --git a/pkg/provider/azure_utils.go b/pkg/provider/azure_utils.go
index 58ac42cbc..cc8daf4d4 100644
--- a/pkg/provider/azure_utils.go
+++ b/pkg/provider/azure_utils.go
@@ -18,16 +18,18 @@ package provider
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"sync"
 
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	"github.com/Azure/go-autorest/autorest/to"
 
 	"k8s.io/klog/v2"
+)
 
-	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
+const (
+	tagsDelimiter        = ","
+	tagKeyValueDelimiter = "="
 )
 
 // lockMap used to lock on entries
@@ -82,6 +84,31 @@ func getContextWithCancel() (context.Context, context.CancelFunc) {
 	return context.WithCancel(context.Background())
 }
 
+// ConvertTagsToMap convert the tags from string to map
+// the valid tags format is "key1=value1,key2=value2", which could be converted to
+// {"key1": "value1", "key2": "value2"}
+func ConvertTagsToMap(tags string) (map[string]string, error) {
+	m := make(map[string]string)
+	if tags == "" {
+		return m, nil
+	}
+	s := strings.Split(tags, tagsDelimiter)
+	for _, tag := range s {
+		kv := strings.Split(tag, tagKeyValueDelimiter)
+		if len(kv) != 2 {
+			return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)
+		}
+		key := strings.TrimSpace(kv[0])
+		if key == "" {
+			return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)
+		}
+		value := strings.TrimSpace(kv[1])
+		m[key] = value
+	}
+
+	return m, nil
+}
+
 func convertMapToMapPointer(origin map[string]string) map[string]*string {
 	newly := make(map[string]*string)
 	for k, v := range origin {
@@ -92,10 +119,10 @@ func convertMapToMapPointer(origin map[string]string) map[string]*string {
 }
 
 func parseTags(tags string) map[string]*string {
-	kvs := strings.Split(tags, consts.TagsDelimiter)
+	kvs := strings.Split(tags, ",")
 	formatted := make(map[string]*string)
 	for _, kv := range kvs {
-		res := strings.Split(kv, consts.TagKeyValueDelimiter)
+		res := strings.Split(kv, "=")
 		if len(res) != 2 {
 			klog.Warningf("parseTags: error when parsing key-value pair %s, would ignore this one", kv)
 			continue
@@ -109,36 +136,3 @@ func parseTags(tags string) map[string]*string {
 	}
 	return formatted
 }
-
-func (az *Cloud) reconcileTags(currentTagsOnResource, newTags map[string]*string) (reconciledTags map[string]*string, changed bool) {
-	var systemTags []string
-	var systemTagsMap sets.String
-
-	if az.SystemTags != "" {
-		systemTags = strings.Split(az.SystemTags, consts.TagsDelimiter)
-		for i := 0; i < len(systemTags); i++ {
-			systemTags[i] = strings.TrimSpace(systemTags[i])
-		}
-		systemTagsMap = sets.NewString(systemTags...)
-	}
-
-	// if the systemTags is not set, just add/update new currentTagsOnResource and not delete old currentTagsOnResource
-	for k, v := range newTags {
-		if vv, ok := currentTagsOnResource[k]; !ok || !strings.EqualFold(to.String(v), to.String(vv)) {
-			currentTagsOnResource[k] = v
-			changed = true
-		}
-	}
-
-	// if the systemTags is set, delete the old currentTagsOnResource
-	if len(systemTagsMap) > 0 {
-		for k := range currentTagsOnResource {
-			if _, ok := newTags[k]; !ok && !systemTagsMap.Has(k) {
-				delete(currentTagsOnResource, k)
-				changed = true
-			}
-		}
-	}
-
-	return currentTagsOnResource, changed
-}
diff --git a/pkg/provider/azure_vmss.go b/pkg/provider/azure_vmss.go
index cb2ea0793..f2516a5d4 100644
--- a/pkg/provider/azure_vmss.go
+++ b/pkg/provider/azure_vmss.go
@@ -81,15 +81,20 @@ type ScaleSet struct {
 // newScaleSet creates a new ScaleSet.
 func newScaleSet(az *Cloud) (VMSet, error) {
 	if az.Config.VmssVirtualMachinesCacheTTLInSeconds == 0 {
-		az.Config.VmssVirtualMachinesCacheTTLInSeconds = vmssVirtualMachinesCacheTTLDefaultInSeconds
+		az.Config.VmssVirtualMachinesCacheTTLInSeconds = consts.VMSSVirtualMachinesCacheTTLDefaultInSeconds
+	}
+
+	var err error
+	as, err := newAvailabilitySet(az)
+	if err != nil {
+		return nil, err
 	}
 	ss := &ScaleSet{
 		Cloud:           az,
-		availabilitySet: newAvailabilitySet(az),
+		availabilitySet: as,
 		vmssVMCache:     &sync.Map{},
 	}
 
-	var err error
 	if !ss.DisableAvailabilitySetNodes {
 		ss.availabilitySetNodesCache, err = ss.newAvailabilitySetNodesCache()
 		if err != nil {
@@ -107,7 +112,7 @@ func newScaleSet(az *Cloud) (VMSet, error) {
 
 func (ss *ScaleSet) getVMSS(vmssName string, crt azcache.AzureCacheReadType) (*compute.VirtualMachineScaleSet, error) {
 	getter := func(vmssName string) (*compute.VirtualMachineScaleSet, error) {
-		cached, err := ss.vmssCache.Get(vmssKey, crt)
+		cached, err := ss.vmssCache.Get(consts.VMSSKey, crt)
 		if err != nil {
 			return nil, err
 		}
@@ -130,7 +135,7 @@ func (ss *ScaleSet) getVMSS(vmssName string, crt azcache.AzureCacheReadType) (*c
 	}
 
 	klog.V(2).Infof("Couldn't find VMSS with name %s, refreshing the cache", vmssName)
-	_ = ss.vmssCache.Delete(vmssKey)
+	_ = ss.vmssCache.Delete(consts.VMSSKey)
 	vmss, err = getter(vmssName)
 	if err != nil {
 		return nil, err
@@ -632,7 +637,7 @@ func (ss *ScaleSet) getNodeIdentityByNodeName(nodeName string, crt azcache.Azure
 			nodeName: nodeName,
 		}
 
-		cached, err := ss.vmssCache.Get(vmssKey, crt)
+		cached, err := ss.vmssCache.Get(consts.VMSSKey, crt)
 		if err != nil {
 			return nil, err
 		}
@@ -915,7 +920,7 @@ func (ss *ScaleSet) getConfigForScaleSetByIPFamily(config *compute.VirtualMachin
 }
 
 // EnsureHostInPool ensures the given VM's Primary NIC's Primary IP Configuration is
-// participating in the specified LoadBalancer Backend Pool, which returns (resourceGroup, vmssName, instanceID, vmssVM, error).
+// participating in the specified LoadBalancer Backend Pool, which returns (resourceGroup, vmasName, instanceID, vmssVM, error).
 func (ss *ScaleSet) EnsureHostInPool(service *v1.Service, nodeName types.NodeName, backendPoolID string, vmSetName string, isInternal bool) (string, string, string, *compute.VirtualMachineScaleSetVM, error) {
 	klog.V(3).Infof("ensuring node %q of scaleset %q in LB backendpool %q", nodeName, vmSetName, backendPoolID)
 	vmName := mapNodeNameToVMName(nodeName)
@@ -1296,7 +1301,7 @@ func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, bac
 }
 
 // ensureBackendPoolDeletedFromNode ensures the loadBalancer backendAddressPools deleted
-// from the specified node, which returns (resourceGroup, vmssName, instanceID, vmssVM, error).
+// from the specified node, which returns (resourceGroup, vmasName, instanceID, vmssVM, error).
 func (ss *ScaleSet) ensureBackendPoolDeletedFromNode(nodeName, backendPoolID string) (string, string, string, *compute.VirtualMachineScaleSetVM, error) {
 	ssName, instanceID, vm, err := ss.getVmssVM(nodeName, azcache.CacheReadTypeDefault)
 	if err != nil {
@@ -1647,13 +1652,13 @@ func (ss *ScaleSet) GetNodeCIDRMasksByProviderID(providerID string) (int, int, e
 	if v4, ok := vmss.Tags[consts.VMSetCIDRIPV4TagKey]; ok && v4 != nil {
 		ipv4Mask, err = strconv.Atoi(to.String(v4))
 		if err != nil {
-			klog.Errorf(" GetNodeCIDRMasksByProviderID: error when paring the value of the ipv4 mask size %s: %v", to.String(v4), err)
+			klog.Errorf("GetNodeCIDRMasksByProviderID: error when paring the value of the ipv4 mask size %s: %v", to.String(v4), err)
 		}
 	}
 	if v6, ok := vmss.Tags[consts.VMSetCIDRIPV6TagKey]; ok && v6 != nil {
 		ipv6Mask, err = strconv.Atoi(to.String(v6))
 		if err != nil {
-			klog.Errorf(" GetNodeCIDRMasksByProviderID: error when paring the value of the ipv6 mask size%s: %v", to.String(v6), err)
+			klog.Errorf("GetNodeCIDRMasksByProviderID: error when paring the value of the ipv6 mask size%s: %v", to.String(v6), err)
 		}
 	}
 
diff --git a/pkg/provider/azure_vmss_cache.go b/pkg/provider/azure_vmss_cache.go
index 3d957ef4f..9bf473a5b 100644
--- a/pkg/provider/azure_vmss_cache.go
+++ b/pkg/provider/azure_vmss_cache.go
@@ -30,17 +30,7 @@ import (
 	"k8s.io/klog/v2"
 
 	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
-)
-
-var (
-	vmssNameSeparator = "_"
-
-	vmssKey                 = "k8svmssKey"
-	availabilitySetNodesKey = "k8sAvailabilitySetNodesKey"
-
-	availabilitySetNodesCacheTTLDefaultInSeconds = 900
-	vmssCacheTTLDefaultInSeconds                 = 600
-	vmssVirtualMachinesCacheTTLDefaultInSeconds  = 600
+	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 )
 
 type vmssVirtualMachinesEntry struct {
@@ -57,14 +47,14 @@ type vmssEntry struct {
 	lastUpdate    time.Time
 }
 
-type availabilitySetEntry struct {
+type availabilitySetNodeEntry struct {
 	vmNames   sets.String
 	nodeNames sets.String
 }
 
 func (ss *ScaleSet) newVMSSCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
-		localCache := &sync.Map{} // [vmssName]*vmssEntry
+		localCache := &sync.Map{} // [vmasName]*vmssEntry
 
 		allResourceGroups, err := ss.GetResourceGroups()
 		if err != nil {
@@ -96,13 +86,13 @@ func (ss *ScaleSet) newVMSSCache() (*azcache.TimedCache, error) {
 	}
 
 	if ss.Config.VmssCacheTTLInSeconds == 0 {
-		ss.Config.VmssCacheTTLInSeconds = vmssCacheTTLDefaultInSeconds
+		ss.Config.VmssCacheTTLInSeconds = consts.VMSSCacheTTLDefaultInSeconds
 	}
 	return azcache.NewTimedcache(time.Duration(ss.Config.VmssCacheTTLInSeconds)*time.Second, getter)
 }
 
 func extractVmssVMName(name string) (string, string, error) {
-	split := strings.SplitAfter(name, vmssNameSeparator)
+	split := strings.SplitAfter(name, consts.VMSSNameSeparator)
 	if len(split) < 2 {
 		klog.V(3).Infof("Failed to extract vmssVMName %q", name)
 		return "", "", ErrorNotVmssInstance
@@ -133,7 +123,7 @@ func (ss *ScaleSet) getVMSSVMCache(resourceGroup, vmssName string) (string, *azc
 
 // gcVMSSVMCache delete stale VMSS VMs caches from deleted VMSSes.
 func (ss *ScaleSet) gcVMSSVMCache() error {
-	cached, err := ss.vmssCache.Get(vmssKey, azcache.CacheReadTypeUnsafe)
+	cached, err := ss.vmssCache.Get(consts.VMSSKey, azcache.CacheReadTypeUnsafe)
 	if err != nil {
 		return err
 	}
@@ -302,7 +292,7 @@ func (ss *ScaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error)
 			return nil, err
 		}
 
-		localCache := availabilitySetEntry{
+		localCache := availabilitySetNodeEntry{
 			vmNames:   vmNames,
 			nodeNames: nodeNames,
 		}
@@ -311,7 +301,7 @@ func (ss *ScaleSet) newAvailabilitySetNodesCache() (*azcache.TimedCache, error)
 	}
 
 	if ss.Config.AvailabilitySetNodesCacheTTLInSeconds == 0 {
-		ss.Config.AvailabilitySetNodesCacheTTLInSeconds = availabilitySetNodesCacheTTLDefaultInSeconds
+		ss.Config.AvailabilitySetNodesCacheTTLInSeconds = consts.AvailabilitySetNodesCacheTTLDefaultInSeconds
 	}
 	return azcache.NewTimedcache(time.Duration(ss.Config.AvailabilitySetNodesCacheTTLInSeconds)*time.Second, getter)
 }
@@ -323,21 +313,21 @@ func (ss *ScaleSet) isNodeManagedByAvailabilitySet(nodeName string, crt azcache.
 		return false, nil
 	}
 
-	cached, err := ss.availabilitySetNodesCache.Get(availabilitySetNodesKey, crt)
+	cached, err := ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, crt)
 	if err != nil {
 		return false, err
 	}
 
-	cachedNodes := cached.(availabilitySetEntry).nodeNames
+	cachedNodes := cached.(availabilitySetNodeEntry).nodeNames
 	// if the node is not in the cache, assume the node has joined after the last cache refresh and attempt to refresh the cache.
 	if !cachedNodes.Has(nodeName) {
 		klog.V(2).Infof("Node %s has joined the cluster since the last VM cache refresh, refreshing the cache", nodeName)
-		cached, err = ss.availabilitySetNodesCache.Get(availabilitySetNodesKey, azcache.CacheReadTypeForceRefresh)
+		cached, err = ss.availabilitySetNodesCache.Get(consts.AvailabilitySetNodesKey, azcache.CacheReadTypeForceRefresh)
 		if err != nil {
 			return false, err
 		}
 	}
 
-	cachedVMs := cached.(availabilitySetEntry).vmNames
+	cachedVMs := cached.(availabilitySetNodeEntry).vmNames
 	return cachedVMs.Has(nodeName), nil
 }


Test patch:
diff --git a/pkg/azureclients/vmasclient/azure_vmasclient_test.go b/pkg/azureclients/vmasclient/azure_vmasclient_test.go
new file mode 100644
index 000000000..703e90ddb
--- /dev/null
+++ b/pkg/azureclients/vmasclient/azure_vmasclient_test.go
@@ -0,0 +1,544 @@
+/*
+Copyright 2021 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 vmasclient
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-12-01/compute"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/to"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/client-go/util/flowcontrol"
+
+	azclients "sigs.k8s.io/cloud-provider-azure/pkg/azureclients"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/armclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/armclient/mockarmclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+const (
+	testResourceID     = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Compute/availabilitySets/vmas1"
+	testResourcePrefix = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Compute/availabilitySets"
+)
+
+func TestNew(t *testing.T) {
+	config := &azclients.ClientConfig{
+		SubscriptionID:          "sub",
+		ResourceManagerEndpoint: "endpoint",
+		Location:                "eastus",
+		RateLimitConfig: &azclients.RateLimitConfig{
+			CloudProviderRateLimit:            true,
+			CloudProviderRateLimitQPS:         0.5,
+			CloudProviderRateLimitBucket:      1,
+			CloudProviderRateLimitQPSWrite:    0.5,
+			CloudProviderRateLimitBucketWrite: 1,
+		},
+		Backoff: &retry.Backoff{Steps: 1},
+	}
+
+	vmasClient := New(config)
+	assert.Equal(t, "sub", vmasClient.subscriptionID)
+	assert.NotEmpty(t, vmasClient.rateLimiterReader)
+	assert.NotEmpty(t, vmasClient.rateLimiterWriter)
+}
+
+func TestNewAzureStack(t *testing.T) {
+	config := &azclients.ClientConfig{
+		CloudName:               "AZURESTACKCLOUD",
+		SubscriptionID:          "sub",
+		ResourceManagerEndpoint: "endpoint",
+		Location:                "eastus",
+		RateLimitConfig: &azclients.RateLimitConfig{
+			CloudProviderRateLimit:            true,
+			CloudProviderRateLimitQPS:         0.5,
+			CloudProviderRateLimitBucket:      1,
+			CloudProviderRateLimitQPSWrite:    0.5,
+			CloudProviderRateLimitBucketWrite: 1,
+		},
+		Backoff: &retry.Backoff{Steps: 1},
+	}
+
+	vmasClient := New(config)
+	assert.Equal(t, "AZURESTACKCLOUD", vmasClient.cloudName)
+	assert.Equal(t, "sub", vmasClient.subscriptionID)
+}
+
+func TestGet(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	expected := compute.AvailabilitySet{Response: autorest.Response{Response: response}}
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Equal(t, expected, result)
+	assert.Nil(t, rerr)
+}
+
+func TestGetNeverRateLimiter(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vmasGetErr := &retry.Error{
+		RawError:  fmt.Errorf("azure cloud provider rate limited(%s) for operation %q", "read", "VMASGet"),
+		Retriable: true,
+	}
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasClient := getTestVMASClientWithNeverRateLimiter(armClient)
+	expected := compute.AvailabilitySet{}
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Equal(t, expected, result)
+	assert.Equal(t, vmasGetErr, rerr)
+}
+
+func TestGetRetryAfterReader(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vmasGetErr := &retry.Error{
+		RawError:   fmt.Errorf("azure cloud provider throttled for operation %s with reason %q", "VMASGet", "client throttled"),
+		Retriable:  true,
+		RetryAfter: getFutureTime(),
+	}
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasClient := getTestVMASClientWithRetryAfterReader(armClient)
+	expected := compute.AvailabilitySet{}
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Equal(t, expected, result)
+	assert.Equal(t, vmasGetErr, rerr)
+}
+
+func TestGetNotFound(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusNotFound,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	expectedVMAS := compute.AvailabilitySet{Response: autorest.Response{}}
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Equal(t, expectedVMAS, result)
+	assert.NotNil(t, rerr)
+	assert.Equal(t, http.StatusNotFound, rerr.HTTPStatusCode)
+}
+
+func TestGetInternalError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusInternalServerError,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	expectedVMAS := compute.AvailabilitySet{Response: autorest.Response{}}
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Equal(t, expectedVMAS, result)
+	assert.NotNil(t, rerr)
+	assert.Equal(t, http.StatusInternalServerError, rerr.HTTPStatusCode)
+}
+
+func TestGetThrottle(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusTooManyRequests,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	throttleErr := &retry.Error{
+		HTTPStatusCode: http.StatusTooManyRequests,
+		RawError:       fmt.Errorf("error"),
+		Retriable:      true,
+		RetryAfter:     time.Unix(100, 0),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourceID, "").Return(response, throttleErr).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.Get(context.TODO(), "rg", "vmas1")
+	assert.Empty(t, result)
+	assert.Equal(t, throttleErr, rerr)
+}
+
+func TestList(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasList := []compute.AvailabilitySet{getTestVMAS("vmas1"), getTestVMAS("vmas2"), getTestVMAS("vmas3")}
+	responseBody, err := json.Marshal(compute.AvailabilitySetListResult{Value: &vmasList})
+	assert.NoError(t, err)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(
+		&http.Response{
+			StatusCode: http.StatusOK,
+			Body:       ioutil.NopCloser(bytes.NewReader(responseBody)),
+		}, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Nil(t, rerr)
+	assert.Equal(t, 3, len(result))
+}
+
+func TestListNotFound(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusNotFound,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	expected := []compute.AvailabilitySet{}
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Equal(t, expected, result)
+	assert.NotNil(t, rerr)
+	assert.Equal(t, http.StatusNotFound, rerr.HTTPStatusCode)
+}
+
+func TestListInternalError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	response := &http.Response{
+		StatusCode: http.StatusInternalServerError,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vmasClient := getTestVMASClient(armClient)
+	expected := []compute.AvailabilitySet{}
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Equal(t, expected, result)
+	assert.NotNil(t, rerr)
+	assert.Equal(t, http.StatusInternalServerError, rerr.HTTPStatusCode)
+}
+
+func TestListThrottle(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	response := &http.Response{
+		StatusCode: http.StatusTooManyRequests,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte("{}"))),
+	}
+	throttleErr := &retry.Error{
+		HTTPStatusCode: http.StatusTooManyRequests,
+		RawError:       fmt.Errorf("error"),
+		Retriable:      true,
+		RetryAfter:     time.Unix(100, 0),
+	}
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(response, throttleErr).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Empty(t, result)
+	assert.NotNil(t, rerr)
+	assert.Equal(t, throttleErr, rerr)
+}
+
+func TestListWithListResponderError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasList := []compute.AvailabilitySet{getTestVMAS("vmas1"), getTestVMAS("vmas2"), getTestVMAS("vmas3")}
+	responseBody, err := json.Marshal(compute.AvailabilitySetListResult{Value: &vmasList})
+	assert.NoError(t, err)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(
+		&http.Response{
+			StatusCode: http.StatusNotFound,
+			Body:       ioutil.NopCloser(bytes.NewReader(responseBody)),
+		}, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.NotNil(t, rerr)
+	assert.Equal(t, 0, len(result))
+}
+
+func TestListWithNextPage(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasList := []compute.AvailabilitySet{getTestVMAS("vmas1"), getTestVMAS("vmas2"), getTestVMAS("vmas3")}
+	partialResponse, err := json.Marshal(compute.AvailabilitySetListResult{Value: &vmasList, NextLink: to.StringPtr("nextLink")})
+	assert.NoError(t, err)
+	pagedResponse, err := json.Marshal(compute.AvailabilitySetListResult{Value: &vmasList})
+	assert.NoError(t, err)
+	armClient.EXPECT().PrepareGetRequest(gomock.Any(), gomock.Any()).Return(&http.Request{}, nil)
+	armClient.EXPECT().Send(gomock.Any(), gomock.Any()).Return(
+		&http.Response{
+			StatusCode: http.StatusOK,
+			Body:       ioutil.NopCloser(bytes.NewReader(pagedResponse)),
+		}, nil)
+	armClient.EXPECT().GetResource(gomock.Any(), testResourcePrefix, "").Return(
+		&http.Response{
+			StatusCode: http.StatusOK,
+			Body:       ioutil.NopCloser(bytes.NewReader(partialResponse)),
+		}, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(2)
+	vmasClient := getTestVMASClient(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Nil(t, rerr)
+	assert.Equal(t, 6, len(result))
+}
+
+func TestListNeverRateLimiter(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vmasListErr := &retry.Error{
+		RawError:  fmt.Errorf("azure cloud provider rate limited(%s) for operation %q", "read", "VMASList"),
+		Retriable: true,
+	}
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasClient := getTestVMASClientWithNeverRateLimiter(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Equal(t, 0, len(result))
+	assert.NotNil(t, rerr)
+	assert.Equal(t, vmasListErr, rerr)
+}
+
+func TestListRetryAfterReader(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vmasListErr := &retry.Error{
+		RawError:   fmt.Errorf("azure cloud provider throttled for operation %s with reason %q", "VMASList", "client throttled"),
+		Retriable:  true,
+		RetryAfter: getFutureTime(),
+	}
+
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	vmasClient := getTestVMASClientWithRetryAfterReader(armClient)
+	result, rerr := vmasClient.List(context.TODO(), "rg")
+	assert.Equal(t, 0, len(result))
+	assert.NotNil(t, rerr)
+	assert.Equal(t, vmasListErr, rerr)
+}
+
+func TestListNextResultsMultiPages(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	tests := []struct {
+		name           string
+		prepareErr     error
+		sendErr        *retry.Error
+		expectedErrMsg string
+	}{
+		{
+			name:       "testlistNextResultsSuccessful",
+			prepareErr: nil,
+			sendErr:    nil,
+		},
+		{
+			name:           "testPrepareGetRequestError",
+			prepareErr:     fmt.Errorf("error"),
+			expectedErrMsg: "Failure preparing next results request",
+		},
+		{
+			name:           "testSendError",
+			sendErr:        &retry.Error{RawError: fmt.Errorf("error")},
+			expectedErrMsg: "Failure sending next results request",
+		},
+	}
+
+	lastResult := compute.AvailabilitySetListResult{
+		NextLink: to.StringPtr("next"),
+	}
+
+	for _, test := range tests {
+		armClient := mockarmclient.NewMockInterface(ctrl)
+		req := &http.Request{
+			Method: "GET",
+		}
+		armClient.EXPECT().PrepareGetRequest(gomock.Any(), gomock.Any()).Return(req, test.prepareErr)
+		if test.prepareErr == nil {
+			armClient.EXPECT().Send(gomock.Any(), req).Return(&http.Response{
+				StatusCode: http.StatusOK,
+				Body:       ioutil.NopCloser(bytes.NewReader([]byte(`{"foo":"bar"}`))),
+			}, test.sendErr)
+			armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any())
+		}
+
+		vmasClient := getTestVMASClient(armClient)
+		result, err := vmasClient.listNextResults(context.TODO(), lastResult)
+		if err != nil {
+			detailedErr := &autorest.DetailedError{}
+			assert.True(t, errors.As(err, detailedErr))
+			assert.Equal(t, detailedErr.Message, test.expectedErrMsg)
+		} else {
+			assert.NoError(t, err)
+		}
+
+		if test.prepareErr != nil {
+			assert.Empty(t, result)
+		} else {
+			assert.NotEmpty(t, result)
+		}
+	}
+}
+
+func TestListNextResultsMultiPagesWithListResponderError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	tests := []struct {
+		name       string
+		prepareErr error
+		sendErr    *retry.Error
+	}{
+		{
+			name:       "testListResponderError",
+			prepareErr: nil,
+			sendErr:    nil,
+		},
+		{
+			name:    "testSendError",
+			sendErr: &retry.Error{RawError: fmt.Errorf("error")},
+		},
+	}
+
+	lastResult := compute.AvailabilitySetListResult{
+		NextLink: to.StringPtr("next"),
+	}
+
+	for _, test := range tests {
+		armClient := mockarmclient.NewMockInterface(ctrl)
+		req := &http.Request{
+			Method: "GET",
+		}
+		armClient.EXPECT().PrepareGetRequest(gomock.Any(), gomock.Any()).Return(req, test.prepareErr)
+		if test.prepareErr == nil {
+			armClient.EXPECT().Send(gomock.Any(), req).Return(&http.Response{
+				StatusCode: http.StatusNotFound,
+				Body:       ioutil.NopCloser(bytes.NewReader([]byte(`{"foo":"bar"}`))),
+			}, test.sendErr)
+			armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any())
+		}
+
+		response := &http.Response{
+			StatusCode: http.StatusNotFound,
+			Body:       ioutil.NopCloser(bytes.NewBuffer([]byte(`{"foo":"bar"}`))),
+		}
+		expected := compute.AvailabilitySetListResult{}
+		expected.Response = autorest.Response{Response: response}
+		vmasClient := getTestVMASClient(armClient)
+		result, err := vmasClient.listNextResults(context.TODO(), lastResult)
+		assert.Error(t, err)
+		if test.sendErr != nil {
+			assert.NotEqual(t, expected, result)
+		} else {
+			assert.Equal(t, expected, result)
+		}
+	}
+}
+
+func getTestVMAS(name string) compute.AvailabilitySet {
+	return compute.AvailabilitySet{
+		ID:       to.StringPtr("/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Compute/availabilitySets/vmas1"),
+		Name:     to.StringPtr(name),
+		Location: to.StringPtr("eastus"),
+		Sku: &compute.Sku{
+			Name:     to.StringPtr("Standard"),
+			Capacity: to.Int64Ptr(3),
+		},
+	}
+}
+
+func getTestVMASClient(armClient armclient.Interface) *Client {
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(&azclients.RateLimitConfig{})
+	return &Client{
+		armClient:         armClient,
+		subscriptionID:    "subscriptionID",
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+	}
+}
+
+func getTestVMASClientWithNeverRateLimiter(armClient armclient.Interface) *Client {
+	rateLimiterReader := flowcontrol.NewFakeNeverRateLimiter()
+	rateLimiterWriter := flowcontrol.NewFakeNeverRateLimiter()
+	return &Client{
+		armClient:         armClient,
+		subscriptionID:    "subscriptionID",
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+	}
+}
+
+func getTestVMASClientWithRetryAfterReader(armClient armclient.Interface) *Client {
+	rateLimiterReader := flowcontrol.NewFakeAlwaysRateLimiter()
+	rateLimiterWriter := flowcontrol.NewFakeAlwaysRateLimiter()
+	return &Client{
+		armClient:         armClient,
+		subscriptionID:    "subscriptionID",
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+		RetryAfterReader:  getFutureTime(),
+		RetryAfterWriter:  getFutureTime(),
+	}
+}
+
+func getFutureTime() time.Time {
+	return time.Unix(3000000000, 0)
+}
diff --git a/pkg/provider/azure_controller_common_test.go b/pkg/provider/azure_controller_common_test.go
index 6bd14f208..f0add60b8 100644
--- a/pkg/provider/azure_controller_common_test.go
+++ b/pkg/provider/azure_controller_common_test.go
@@ -85,26 +85,11 @@ func TestCommonAttachDisk(t *testing.T) {
 				DiskProperties: &compute.DiskProperties{
 					Encryption: &compute.Encryption{DiskEncryptionSetID: &diskEncryptionSetID, Type: compute.EncryptionTypeEncryptionAtRestWithCustomerKey},
 					DiskSizeGB: to.Int32Ptr(4096),
-					DiskState:  compute.Unattached,
 				},
 				Tags: testTags},
 			expectedLun: 1,
 			expectedErr: false,
 		},
-		{
-			desc:     "an error shall be returned if disk state is not Unattached",
-			vmList:   map[string]string{"vm1": "PowerState/Running"},
-			nodeName: "vm1",
-			existedDisk: compute.Disk{Name: to.StringPtr("disk-name"),
-				DiskProperties: &compute.DiskProperties{
-					Encryption: &compute.Encryption{DiskEncryptionSetID: &diskEncryptionSetID, Type: compute.EncryptionTypeEncryptionAtRestWithCustomerKey},
-					DiskSizeGB: to.Int32Ptr(4096),
-					DiskState:  compute.Attached,
-				},
-				Tags: testTags},
-			expectedLun: -1,
-			expectedErr: true,
-		},
 		{
 			desc:         "an error shall be returned if there's invalid disk uri",
 			vmList:       map[string]string{"vm1": "PowerState/Running"},
diff --git a/pkg/provider/azure_instances_test.go b/pkg/provider/azure_instances_test.go
index a51711b88..d406ba746 100644
--- a/pkg/provider/azure_instances_test.go
+++ b/pkg/provider/azure_instances_test.go
@@ -200,7 +200,7 @@ func TestInstanceID(t *testing.T) {
 		if test.nilVMSet {
 			cloud.VMSet = nil
 		} else {
-			cloud.VMSet = newAvailabilitySet(cloud)
+			cloud.VMSet, _ = newAvailabilitySet(cloud)
 		}
 		cloud.Config.VMType = test.vmType
 		cloud.Config.UseInstanceMetadata = test.useInstanceMetadata
@@ -561,7 +561,7 @@ func TestNodeAddresses(t *testing.T) {
 		if test.nilVMSet {
 			cloud.VMSet = nil
 		} else {
-			cloud.VMSet = newAvailabilitySet(cloud)
+			cloud.VMSet, _ = newAvailabilitySet(cloud)
 		}
 		cloud.Config.VMType = test.vmType
 		cloud.Config.UseInstanceMetadata = test.useInstanceMetadata
diff --git a/pkg/provider/azure_loadbalancer_test.go b/pkg/provider/azure_loadbalancer_test.go
index c77163022..4c4e72bca 100644
--- a/pkg/provider/azure_loadbalancer_test.go
+++ b/pkg/provider/azure_loadbalancer_test.go
@@ -3767,48 +3767,27 @@ func buildDefaultTestLB(name string, backendIPConfigs []string) network.LoadBala
 }
 
 func TestEnsurePIPTagged(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	cloud := GetTestCloud(ctrl)
-	cloud.Tags = "a=x,y=z"
+	t.Run("ensurePIPTagged should ensure the pip is tagged as configured", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+		cloud := GetTestCloud(ctrl)
+		cloud.Tags = "a=x,y=z"
 
-	service := v1.Service{
-		ObjectMeta: metav1.ObjectMeta{
-			Annotations: map[string]string{
-				consts.ServiceAnnotationAzurePIPTags: "a=b,c=d,e=,=f,ghi",
+		service := v1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					consts.ServiceAnnotationAzurePIPTags: "a=b,c=d,e=,=f,ghi",
+				},
 			},
-		},
-	}
-	pip := network.PublicIPAddress{
-		Tags: map[string]*string{
-			consts.ClusterNameKey: to.StringPtr("testCluster"),
-			consts.ServiceTagKey:  to.StringPtr("default/svc1,default/svc2"),
-			"foo":                 to.StringPtr("bar"),
-			"a":                   to.StringPtr("j"),
-			"m":                   to.StringPtr("n"),
-		},
-	}
-
-	t.Run("ensurePIPTagged should ensure the pip is tagged as configured", func(t *testing.T) {
-		expectedPIP := network.PublicIPAddress{
+		}
+		pip := network.PublicIPAddress{
 			Tags: map[string]*string{
 				consts.ClusterNameKey: to.StringPtr("testCluster"),
 				consts.ServiceTagKey:  to.StringPtr("default/svc1,default/svc2"),
 				"foo":                 to.StringPtr("bar"),
-				"a":                   to.StringPtr("b"),
-				"c":                   to.StringPtr("d"),
-				"y":                   to.StringPtr("z"),
-				"m":                   to.StringPtr("n"),
+				"a":                   to.StringPtr("j"),
 			},
 		}
-		changed := cloud.ensurePIPTagged(&service, &pip)
-		assert.True(t, changed)
-		assert.Equal(t, expectedPIP, pip)
-	})
-
-	t.Run("ensurePIPTagged should delete the old tags if tht SystemTags is set", func(t *testing.T) {
-		cloud.SystemTags = "a,foo"
 		expectedPIP := network.PublicIPAddress{
 			Tags: map[string]*string{
 				consts.ClusterNameKey: to.StringPtr("testCluster"),
@@ -3825,45 +3804,6 @@ func TestEnsurePIPTagged(t *testing.T) {
 	})
 }
 
-func TestEnsureLoadBalancerTagged(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	for _, tc := range []struct {
-		description               string
-		existedTags, expectedTags map[string]*string
-		newTags, systemTags       string
-		expectedChanged           bool
-	}{
-		{
-			description:     "ensureLoadBalancerTagged should not delete the old tags if SystemTags is not specified",
-			existedTags:     map[string]*string{"a": to.StringPtr("b")},
-			newTags:         "c=d",
-			expectedTags:    map[string]*string{"a": to.StringPtr("b"), "c": to.StringPtr("d")},
-			expectedChanged: true,
-		},
-		{
-			description:     "ensureLoadBalancerTagged should delete the old tags if SystemTags is specified",
-			existedTags:     map[string]*string{"a": to.StringPtr("b"), "c": to.StringPtr("d"), "h": to.StringPtr("i")},
-			newTags:         "c=e,f=g",
-			systemTags:      "a,x,y,z",
-			expectedTags:    map[string]*string{"a": to.StringPtr("b"), "c": to.StringPtr("e"), "f": to.StringPtr("g")},
-			expectedChanged: true,
-		},
-	} {
-		t.Run(tc.description, func(t *testing.T) {
-			cloud := GetTestCloud(ctrl)
-			cloud.Tags = tc.newTags
-			cloud.SystemTags = tc.systemTags
-			lb := &network.LoadBalancer{Tags: tc.existedTags}
-
-			changed := cloud.ensureLoadBalancerTagged(lb)
-			assert.Equal(t, tc.expectedChanged, changed)
-			assert.Equal(t, tc.expectedTags, lb.Tags)
-		})
-	}
-}
-
 func TestShouldChangeLoadBalancer(t *testing.T) {
 	t.Run("shouldChangeLoadBalancer should return true if the mode is different from the current vm set", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
diff --git a/pkg/provider/azure_standard_test.go b/pkg/provider/azure_standard_test.go
index d95de21e7..7cdd33d96 100644
--- a/pkg/provider/azure_standard_test.go
+++ b/pkg/provider/azure_standard_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package provider
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -35,7 +36,9 @@ import (
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/interfaceclient/mockinterfaceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/publicipclient/mockpublicipclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmasclient/mockvmasclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/vmclient/mockvmclient"
+	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"
 )
@@ -1752,3 +1755,150 @@ func TestStandardGetNodeNameByIPConfigurationID(t *testing.T) {
 	assert.Equal(t, "k8s-agentpool1-00000000-0", nodeName)
 	assert.Equal(t, "agentpool1-availabilityset-00000000", asName)
 }
+
+func TestGetAvailabilitySetByNodeName(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		description   string
+		nodeName      string
+		vmasVMIDs     []string
+		vmasListError *retry.Error
+		expectedErr   error
+	}{
+		{
+			description: "getAvailabilitySetByNodeName should return the correct VMAS",
+			nodeName:    "vm-1",
+			vmasVMIDs:   []string{"/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-1"},
+		},
+		{
+			description: "getAvailabilitySetByNodeName should return cloudprovider.InstanceNotFound if there's no matching VMAS",
+			nodeName:    "vm-2",
+			vmasVMIDs:   []string{"/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-1"},
+			expectedErr: cloudprovider.InstanceNotFound,
+		},
+		{
+			description:   "getAvailabilitySetByNodeName should report an error if there's something wrong during an api call",
+			nodeName:      "vm-1",
+			vmasVMIDs:     []string{"/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-1"},
+			vmasListError: &retry.Error{RawError: fmt.Errorf("error during vmas list")},
+			expectedErr:   fmt.Errorf("Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error during vmas list"),
+		},
+		{
+			description: "getAvailabilitySetByNodeName should report an error if the vmID on the vmas is invalid",
+			nodeName:    "vm-1",
+			vmasVMIDs:   []string{"invalid"},
+			expectedErr: fmt.Errorf("invalid vm ID invalid"),
+		},
+	}
+
+	for _, test := range testCases {
+		cloud := GetTestCloud(ctrl)
+		vmSet, err := newAvailabilitySet(cloud)
+		assert.NoError(t, err)
+		as := vmSet.(*availabilitySet)
+
+		mockVMASClient := mockvmasclient.NewMockInterface(ctrl)
+		cloud.AvailabilitySetsClient = mockVMASClient
+
+		subResources := make([]compute.SubResource, 0)
+		for _, vmID := range test.vmasVMIDs {
+			subResources = append(subResources, compute.SubResource{
+				ID: to.StringPtr(vmID),
+			})
+		}
+		expected := compute.AvailabilitySet{
+			Name: to.StringPtr("vmas-1"),
+			AvailabilitySetProperties: &compute.AvailabilitySetProperties{
+				VirtualMachines: &subResources,
+			},
+		}
+		mockVMASClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.AvailabilitySet{expected}, test.vmasListError).AnyTimes()
+
+		actual, err := as.getAvailabilitySetByNodeName(test.nodeName, azcache.CacheReadTypeDefault)
+		if test.expectedErr != nil {
+			assert.EqualError(t, test.expectedErr, err.Error(), test.description)
+		}
+		if actual != nil {
+			assert.Equal(t, expected, *actual, test.description)
+		}
+	}
+}
+
+func TestGetNodeCIDRMasksByProviderIDAvailabilitySet(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	for _, tc := range []struct {
+		description, providerID                    string
+		tags                                       map[string]*string
+		expectedIPV4MaskSize, expectedIPV6MaskSize int
+		expectedErr                                error
+	}{
+		{
+			description: "GetNodeCIDRMasksByProviderID should report an error if the providerID is not valid",
+			providerID:  "invalid",
+			expectedErr: errors.New("error splitting providerID"),
+		},
+		{
+			description: "GetNodeCIDRMaksByProviderID should return the correct mask sizes",
+			providerID:  "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-0",
+			tags: map[string]*string{
+				consts.VMSetCIDRIPV4TagKey: to.StringPtr("24"),
+				consts.VMSetCIDRIPV6TagKey: to.StringPtr("64"),
+			},
+			expectedIPV4MaskSize: 24,
+			expectedIPV6MaskSize: 64,
+		},
+		{
+			description:          "GetNodeCIDRMaksByProviderID should report cloudprovider.InstanceNotFound if there is no matching vmas",
+			providerID:           "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-1",
+			expectedIPV4MaskSize: 24,
+			expectedIPV6MaskSize: 64,
+		},
+		{
+			description: "GetNodeCIDRMaksByProviderID should return the correct mask sizes even if some of the tags are not specified",
+			providerID:  "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-0",
+			tags: map[string]*string{
+				consts.VMSetCIDRIPV4TagKey: to.StringPtr("24"),
+			},
+			expectedIPV4MaskSize: 24,
+		},
+		{
+			description: "GetNodeCIDRMaksByProviderID should not fail even if some of the tag is invalid",
+			providerID:  "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-0",
+			tags: map[string]*string{
+				consts.VMSetCIDRIPV4TagKey: to.StringPtr("abc"),
+				consts.VMSetCIDRIPV6TagKey: to.StringPtr("64"),
+			},
+			expectedIPV6MaskSize: 64,
+		},
+	} {
+		t.Run(tc.description, func(t *testing.T) {
+			cloud := GetTestCloud(ctrl)
+			vmSet, err := newAvailabilitySet(cloud)
+			assert.NoError(t, err)
+			as := vmSet.(*availabilitySet)
+
+			mockVMASClient := mockvmasclient.NewMockInterface(ctrl)
+			cloud.AvailabilitySetsClient = mockVMASClient
+
+			expected := compute.AvailabilitySet{
+				Name: to.StringPtr("vmas-1"),
+				AvailabilitySetProperties: &compute.AvailabilitySetProperties{
+					VirtualMachines: &[]compute.SubResource{
+						{ID: to.StringPtr("/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm-0")},
+					},
+				},
+				Tags: tc.tags,
+			}
+			mockVMASClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.AvailabilitySet{expected}, nil).AnyTimes()
+
+			ipv4MaskSize, ipv6MaskSize, err := as.GetNodeCIDRMasksByProviderID(tc.providerID)
+			assert.Equal(t, tc.expectedErr, err)
+			assert.Equal(t, tc.expectedIPV4MaskSize, ipv4MaskSize)
+			assert.Equal(t, tc.expectedIPV6MaskSize, ipv6MaskSize)
+		})
+	}
+}
diff --git a/pkg/provider/azure_utils_test.go b/pkg/provider/azure_utils_test.go
index f42f4552c..203959383 100644
--- a/pkg/provider/azure_utils_test.go
+++ b/pkg/provider/azure_utils_test.go
@@ -17,8 +17,11 @@ limitations under the License.
 package provider
 
 import (
+	"reflect"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestSimpleLockEntry(t *testing.T) {
@@ -81,3 +84,69 @@ func ensureNoCallback(t *testing.T, callbackChan <-chan interface{}) bool {
 		return true
 	}
 }
+
+func TestConvertTagsToMap(t *testing.T) {
+	testCases := []struct {
+		desc           string
+		tags           string
+		expectedOutput map[string]string
+		expectedError  bool
+	}{
+		{
+			desc:           "should return empty map when tag is empty",
+			tags:           "",
+			expectedOutput: map[string]string{},
+			expectedError:  false,
+		},
+		{
+			desc: "sing valid tag should be converted",
+			tags: "key=value",
+			expectedOutput: map[string]string{
+				"key": "value",
+			},
+			expectedError: false,
+		},
+		{
+			desc: "multiple valid tags should be converted",
+			tags: "key1=value1,key2=value2",
+			expectedOutput: map[string]string{
+				"key1": "value1",
+				"key2": "value2",
+			},
+			expectedError: false,
+		},
+		{
+			desc: "whitespaces should be trimmed",
+			tags: "key1=value1, key2=value2",
+			expectedOutput: map[string]string{
+				"key1": "value1",
+				"key2": "value2",
+			},
+			expectedError: false,
+		},
+		{
+			desc:           "should return error for invalid format",
+			tags:           "foo,bar",
+			expectedOutput: nil,
+			expectedError:  true,
+		},
+		{
+			desc:           "should return error for when key is missed",
+			tags:           "key1=value1,=bar",
+			expectedOutput: nil,
+			expectedError:  true,
+		},
+	}
+
+	for i, c := range testCases {
+		m, err := ConvertTagsToMap(c.tags)
+		if c.expectedError {
+			assert.NotNil(t, err, "TestCase[%d]: %s", i, c.desc)
+		} else {
+			assert.Nil(t, err, "TestCase[%d]: %s", i, c.desc)
+			if !reflect.DeepEqual(m, c.expectedOutput) {
+				t.Errorf("got: %v, expected: %v, desc: %v", m, c.expectedOutput, c.desc)
+			}
+		}
+	}
+}
diff --git a/pkg/provider/azure_vmss_test.go b/pkg/provider/azure_vmss_test.go
index 0334cfa8f..9fdca2d2c 100644
--- a/pkg/provider/azure_vmss_test.go
+++ b/pkg/provider/azure_vmss_test.go
@@ -2479,7 +2479,7 @@ func TestGetNodeCIDRMasksByProviderID(t *testing.T) {
 			expectedIPV4MaskSize: 24,
 		},
 		{
-			description: "GetNodeCIDRMaksByProviderID should fail even if some of the tag is invalid",
+			description: "GetNodeCIDRMaksByProviderID should not fail even if some of the tag is invalid",
 			providerID:  "azure:///subscriptions/sub/resourceGroups/rg1/providers/Microsoft.Compute/virtualMachineScaleSets/vmss/virtualMachines/0",
 			tags: map[string]*string{
 				consts.VMSetCIDRIPV4TagKey: to.StringPtr("abc"),