Repository: kubernetes-sigs/cloud-provider-azure
Pull request #1732 (2022-05-16 04:14:48)

[release-1.1]refactor: privateendpointclient and virtualnetworklinkclient change
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?
/kind cleanup
<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature
/kind design

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:
Re-implement privateEndpointClient and virtualNetworkLinkClient to use armclient. This can facilitate ARG integration in the near future.
Cherry pick into release-1.1
#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #
related #1576 
#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
Azure privateEndpointClient APIs are changed:
1. Get() and CreateOrUpdate() return *retry.Error instead of error
2. Add etag argument for CreateOrUpdate() function.
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```



Patch:
diff --git a/pkg/azureclients/armclient/azure_armclient.go b/pkg/azureclients/armclient/azure_armclient.go
index 907072e13..764d2e0a0 100644
--- a/pkg/azureclients/armclient/azure_armclient.go
+++ b/pkg/azureclients/armclient/azure_armclient.go
@@ -492,7 +492,7 @@ func (c *Client) PutResourcesInBatches(ctx context.Context, resources map[string
 	return responses
 }
 
-// PutResourceWithDecorators puts a resource by resource ID
+// PutResourceWithDecorators puts a resource by resource ID and waits for response
 func (c *Client) PutResourceWithDecorators(ctx context.Context, resourceID string, parameters interface{}, decorators []autorest.PrepareDecorator) (*http.Response, *retry.Error) {
 	return c.putResourceWithDecorators(ctx, resourceID, parameters, decorators, true)
 }
diff --git a/pkg/azureclients/privateendpointclient/azure_privateendpointclient.go b/pkg/azureclients/privateendpointclient/azure_privateendpointclient.go
index 15096d412..3a90a2537 100644
--- a/pkg/azureclients/privateendpointclient/azure_privateendpointclient.go
+++ b/pkg/azureclients/privateendpointclient/azure_privateendpointclient.go
@@ -14,53 +14,215 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+// Package privateendpointclient implements the client for Private Endpoint.
 package privateendpointclient
 
 import (
 	"context"
+	"net/http"
+	"strings"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/azure"
+
+	"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{}
 
+const (
+	PEResourceType = "Microsoft.Network/privateEndpoints"
+)
+
 // Client implements privateendpointclient Interface.
 type Client struct {
-	privateEndpointClient network.PrivateEndpointsClient
+	armClient      armclient.Interface
+	cloudName      string
+	subscriptionID string
+
+	// Rate limiting configures.
+	rateLimiterReader flowcontrol.RateLimiter
+	rateLimiterWriter flowcontrol.RateLimiter
+
+	// ARM throttling configures.
+	RetryAfterReader time.Time
+	RetryAfterWriter time.Time
 }
 
 // New creates a new private endpoint client.
 func New(config *azclients.ClientConfig) *Client {
-	privateEndpointClient := network.NewPrivateEndpointsClient(config.SubscriptionID)
-	privateEndpointClient.Authorizer = config.Authorizer
+	apiVersion := APIVersion
+	if strings.EqualFold(config.CloudName, AzureStackCloudName) && !config.DisableAzureStackCloud {
+		// To check whether Azure Stack Cloud supports a resource and the supported API version, refer to:
+		// https://docs.microsoft.com/en-us/azure-stack/user/azure-stack-profiles-azure-resource-manager-versions?view=azs-2108
+		klog.Warningf("Azure Stack is not supported for Private Endpoint API")
+	}
+	armClient := armclient.New(config.Authorizer, *config, config.ResourceManagerEndpoint, apiVersion)
+
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(config.RateLimitConfig)
+	if azclients.RateLimitEnabled(config.RateLimitConfig) {
+		klog.V(2).Infof("Azure PrivateEndpointsClient (read ops) using rate limit config: QPS=%g, bucket=%d",
+			config.RateLimitConfig.CloudProviderRateLimitQPS,
+			config.RateLimitConfig.CloudProviderRateLimitBucket)
+		klog.V(2).Infof("Azure PrivateEndpointsClient (write ops) using rate limit config: QPS=%g, bucket=%d",
+			config.RateLimitConfig.CloudProviderRateLimitQPSWrite,
+			config.RateLimitConfig.CloudProviderRateLimitBucketWrite)
+	}
 
 	client := &Client{
-		privateEndpointClient: privateEndpointClient,
+		armClient:         armClient,
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+		subscriptionID:    config.SubscriptionID,
+		cloudName:         config.CloudName,
 	}
 	return client
 }
 
 // CreateOrUpdate creates or updates a private endpoint.
-func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, endpointName string, privateEndpoint network.PrivateEndpoint, waitForCompletion bool) error {
-	createOrUpdateFuture, err := c.privateEndpointClient.CreateOrUpdate(ctx, resourceGroupName, endpointName, privateEndpoint)
-	if err != nil {
-		klog.V(5).Infof("Received error for %s, resourceGroup: %s, error: %s", "privateendpoint.put.request", resourceGroupName, err)
-		return err
+func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, endpointName string, privateEndpoint network.PrivateEndpoint, etag string, waitForCompletion bool) *retry.Error {
+	mc := metrics.NewMetricContext("private_endpoints", "create_or_update", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return retry.GetRateLimitError(true, "PrivateEndpointCreateOrUpdate")
 	}
-	if waitForCompletion {
-		err = createOrUpdateFuture.WaitForCompletionRef(ctx, c.privateEndpointClient.Client)
-		if err != nil {
-			klog.V(5).Infof("Received error while waiting for completion for %s, resourceGroup: %s, error: %s", "privateendpoint.put.request", resourceGroupName, err)
-			return err
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("PrivateEndpointCreateOrUpdate", "client throttled", c.RetryAfterWriter)
+		return rerr
+	}
+
+	rerr := c.createOrUpdatePE(ctx, resourceGroupName, endpointName, privateEndpoint, etag, waitForCompletion)
+	mc.Observe(rerr)
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterWriter = rerr.RetryAfter
 		}
 
+		return rerr
 	}
 	return nil
 }
 
+func (c *Client) createOrUpdatePE(ctx context.Context, resourceGroupName string, endpointName string, parameters network.PrivateEndpoint, etag string, waitForCompletion bool) *retry.Error {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		PEResourceType,
+		endpointName,
+	)
+	decorators := []autorest.PrepareDecorator{
+		autorest.WithPathParameters("{resourceID}", map[string]interface{}{"resourceID": resourceID}),
+		autorest.WithJSON(parameters),
+	}
+	if etag != "" {
+		decorators = append(decorators, autorest.WithHeader("If-Match", autorest.String(etag)))
+	}
+
+	var response *http.Response
+	var rerr *retry.Error
+	if waitForCompletion {
+		response, rerr = c.armClient.PutResourceWithDecorators(ctx, resourceID, parameters, decorators)
+	} else {
+		response, rerr = c.armClient.PutResourceWithDecoratorsAsync(ctx, resourceID, parameters, decorators)
+	}
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "privateendpoint.put.request", resourceID, rerr.Error())
+		return rerr
+	}
+
+	if !waitForCompletion {
+		return nil
+	}
+
+	if response != nil && response.StatusCode != http.StatusNoContent {
+		_, rerr = c.createOrUpdateResponder(response)
+		if rerr != nil {
+			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "privateendpoint.put.request", resourceID, rerr.Error())
+			return rerr
+		}
+	}
+	return nil
+}
+
+func (c *Client) createOrUpdateResponder(resp *http.Response) (*network.PrivateEndpoint, *retry.Error) {
+	result := &network.PrivateEndpoint{}
+	err := autorest.Respond(
+		resp,
+		azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusCreated),
+		autorest.ByUnmarshallingJSON(&result))
+	result.Response = autorest.Response{Response: resp}
+	return result, retry.GetError(resp, err)
+}
+
 // Get gets the private endpoint
-func (c *Client) Get(ctx context.Context, resourceGroupName string, privateEndpointName string, expand string) (result network.PrivateEndpoint, err error) {
-	return c.privateEndpointClient.Get(ctx, resourceGroupName, privateEndpointName, expand)
+func (c *Client) Get(ctx context.Context, resourceGroupName string, privateEndpointName string, expand string) (network.PrivateEndpoint, *retry.Error) {
+	mc := metrics.NewMetricContext("private_endpoints", "get", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterReader.TryAccept() {
+		mc.RateLimitedCount()
+		return network.PrivateEndpoint{}, retry.GetRateLimitError(false, "PrivateEndpointGet")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterReader.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("PrivateEndpointGet", "client throttled", c.RetryAfterReader)
+		return network.PrivateEndpoint{}, rerr
+	}
+	result, rerr := c.getPE(ctx, resourceGroupName, privateEndpointName, expand)
+
+	mc.Observe(rerr)
+	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
+}
+
+// getPE gets a private endpoint.
+func (c *Client) getPE(ctx context.Context, resourceGroupName string, privateEndpointName string, expand string) (network.PrivateEndpoint, *retry.Error) {
+	resourceID := armclient.GetResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		PEResourceType,
+		privateEndpointName,
+	)
+	result := network.PrivateEndpoint{}
+	response, rerr := c.armClient.GetResourceWithExpandQuery(ctx, resourceID, expand)
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "privateendpoint.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", "privateendpoint.get.respond", resourceID, err)
+		return result, retry.GetError(response, err)
+	}
+
+	result.Response = autorest.Response{Response: response}
+	return result, nil
 }
diff --git a/pkg/azureclients/privateendpointclient/doc.go b/pkg/azureclients/privateendpointclient/doc.go
new file mode 100644
index 000000000..b01f75b2e
--- /dev/null
+++ b/pkg/azureclients/privateendpointclient/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2022 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 privateendpointclient implements the client for PrivateEndpoint.
+package privateendpointclient // import "sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privateendpointclient"
diff --git a/pkg/azureclients/privateendpointclient/interface.go b/pkg/azureclients/privateendpointclient/interface.go
index 8aa1d4cd1..8142ef458 100644
--- a/pkg/azureclients/privateendpointclient/interface.go
+++ b/pkg/azureclients/privateendpointclient/interface.go
@@ -20,6 +20,15 @@ import (
 	"context"
 
 	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+const (
+	// APIVersion is the API version for network.
+	APIVersion = "2021-02-01"
+	// AzureStackCloudName is the cloud name of Azure Stack
+	AzureStackCloudName = "AZURESTACKCLOUD"
 )
 
 // Interface is the client interface for Private Endpoints.
@@ -27,8 +36,8 @@ import (
 type Interface interface {
 
 	// Get gets the private endpoint
-	Get(ctx context.Context, resourceGroupName string, privateEndpointName string, expand string) (result network.PrivateEndpoint, err error)
+	Get(ctx context.Context, resourceGroupName string, privateEndpointName string, expand string) (result network.PrivateEndpoint, rerr *retry.Error)
 
 	// CreateOrUpdate creates or updates a private endpoint.
-	CreateOrUpdate(ctx context.Context, resourceGroupName string, endpointName string, privateEndpoint network.PrivateEndpoint, waitForCompletion bool) error
+	CreateOrUpdate(ctx context.Context, resourceGroupName string, endpointName string, privateEndpoint network.PrivateEndpoint, etag string, waitForCompletion bool) *retry.Error
 }
diff --git a/pkg/azureclients/privateendpointclient/mockprivateendpointclient/doc.go b/pkg/azureclients/privateendpointclient/mockprivateendpointclient/doc.go
index 2aafee6c8..9063c82b8 100644
--- a/pkg/azureclients/privateendpointclient/mockprivateendpointclient/doc.go
+++ b/pkg/azureclients/privateendpointclient/mockprivateendpointclient/doc.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Kubernetes Authors.
+Copyright 2022 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.
diff --git a/pkg/azureclients/privateendpointclient/mockprivateendpointclient/interface.go b/pkg/azureclients/privateendpointclient/mockprivateendpointclient/interface.go
index 9be0505fa..4719fe94d 100644
--- a/pkg/azureclients/privateendpointclient/mockprivateendpointclient/interface.go
+++ b/pkg/azureclients/privateendpointclient/mockprivateendpointclient/interface.go
@@ -27,6 +27,7 @@ import (
 
 	network "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
 	gomock "github.com/golang/mock/gomock"
+	retry "sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
 // MockInterface is a mock of Interface interface.
@@ -53,25 +54,25 @@ func (m *MockInterface) EXPECT() *MockInterfaceMockRecorder {
 }
 
 // CreateOrUpdate mocks base method.
-func (m *MockInterface) CreateOrUpdate(ctx context.Context, resourceGroupName, endpointName string, privateEndpoint network.PrivateEndpoint, waitForCompletion bool) error {
+func (m *MockInterface) CreateOrUpdate(ctx context.Context, resourceGroupName, endpointName string, privateEndpoint network.PrivateEndpoint, etag string, waitForCompletion bool) *retry.Error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateOrUpdate", ctx, resourceGroupName, endpointName, privateEndpoint, waitForCompletion)
-	ret0, _ := ret[0].(error)
+	ret := m.ctrl.Call(m, "CreateOrUpdate", ctx, resourceGroupName, endpointName, privateEndpoint, etag, waitForCompletion)
+	ret0, _ := ret[0].(*retry.Error)
 	return ret0
 }
 
 // CreateOrUpdate indicates an expected call of CreateOrUpdate.
-func (mr *MockInterfaceMockRecorder) CreateOrUpdate(ctx, resourceGroupName, endpointName, privateEndpoint, waitForCompletion interface{}) *gomock.Call {
+func (mr *MockInterfaceMockRecorder) CreateOrUpdate(ctx, resourceGroupName, endpointName, privateEndpoint, etag, waitForCompletion interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateOrUpdate", reflect.TypeOf((*MockInterface)(nil).CreateOrUpdate), ctx, resourceGroupName, endpointName, privateEndpoint, waitForCompletion)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateOrUpdate", reflect.TypeOf((*MockInterface)(nil).CreateOrUpdate), ctx, resourceGroupName, endpointName, privateEndpoint, etag, waitForCompletion)
 }
 
 // Get mocks base method.
-func (m *MockInterface) Get(ctx context.Context, resourceGroupName, privateEndpointName, expand string) (network.PrivateEndpoint, error) {
+func (m *MockInterface) Get(ctx context.Context, resourceGroupName, privateEndpointName, expand string) (network.PrivateEndpoint, *retry.Error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Get", ctx, resourceGroupName, privateEndpointName, expand)
 	ret0, _ := ret[0].(network.PrivateEndpoint)
-	ret1, _ := ret[1].(error)
+	ret1, _ := ret[1].(*retry.Error)
 	return ret0, ret1
 }
 
diff --git a/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient.go b/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient.go
index 2c0befbc3..46d9239b0 100644
--- a/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient.go
+++ b/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Kubernetes Authors.
+Copyright 2022 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.
@@ -18,48 +18,212 @@ package virtualnetworklinksclient
 
 import (
 	"context"
+	"net/http"
+	"strings"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/privatedns/mgmt/2018-09-01/privatedns"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/azure"
+	"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{}
 
+const (
+	privateDNSZoneResourceType     = "Microsoft.Network/privateDnsZones"
+	virtualNetworkLinkResourceType = "virtualNetworkLinks"
+)
+
 // Client implements virtualnetworklinksclient Interface.
 type Client struct {
-	virtualNetworkLinksClient privatedns.VirtualNetworkLinksClient
+	armClient      armclient.Interface
+	cloudName      string
+	subscriptionID string
+
+	// Rate limiting configures.
+	rateLimiterReader flowcontrol.RateLimiter
+	rateLimiterWriter flowcontrol.RateLimiter
+
+	// ARM throttling configures.
+	RetryAfterReader time.Time
+	RetryAfterWriter time.Time
 }
 
 // New creates a new virtualnetworklinks client.
 func New(config *azclients.ClientConfig) *Client {
-	virtualNetworkLinksClient := privatedns.NewVirtualNetworkLinksClient(config.SubscriptionID)
-	virtualNetworkLinksClient.Authorizer = config.Authorizer
+	apiVersion := APIVersion
+	if strings.EqualFold(config.CloudName, AzureStackCloudName) && !config.DisableAzureStackCloud {
+		klog.Warningf("Azure Stack is not supported for Virtual Network Link API")
+	}
+	armClient := armclient.New(config.Authorizer, *config, config.ResourceManagerEndpoint, apiVersion)
+
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(config.RateLimitConfig)
+	if azclients.RateLimitEnabled(config.RateLimitConfig) {
+		klog.V(2).Infof("Azure VirtualNetworkLinksClient (read ops) using rate limit config: QPS=%g, bucket=%d",
+			config.RateLimitConfig.CloudProviderRateLimitQPS,
+			config.RateLimitConfig.CloudProviderRateLimitBucket)
+		klog.V(2).Infof("Azure VirtualNetworkLinksClient (write ops) using rate limit config: QPS=%g, bucket=%d",
+			config.RateLimitConfig.CloudProviderRateLimitQPSWrite,
+			config.RateLimitConfig.CloudProviderRateLimitBucketWrite)
+	}
 
 	client := &Client{
-		virtualNetworkLinksClient: virtualNetworkLinksClient,
+		armClient:         armClient,
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+		subscriptionID:    config.SubscriptionID,
+		cloudName:         config.CloudName,
 	}
 	return client
 }
 
 // CreateOrUpdate creates or updates a virtual network link
-func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, waitForCompletion bool) error {
-	createOrUpdateFuture, err := c.virtualNetworkLinksClient.CreateOrUpdate(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, "", "*")
-	if err != nil {
-		klog.V(5).Infof("Received error for %s, resourceGroup: %s, privateZoneName: %s, error: %s", "virtualnetworklinks.put.request", resourceGroupName, privateZoneName, err)
-		return err
+func (c *Client) CreateOrUpdate(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, etag string, waitForCompletion bool) *retry.Error {
+	mc := metrics.NewMetricContext("virtual_network_links", "create_or_update", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return retry.GetRateLimitError(true, "VirtualNetworkLinkCreateOrUpdate")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("VirtualNetworkLinkCreateOrUpdate", "client throttled", c.RetryAfterWriter)
+		return rerr
 	}
+
+	rerr := c.createOrUpdateVirtualNetworkLink(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, etag, waitForCompletion)
+	mc.Observe(rerr)
+	if rerr != nil {
+		if rerr.IsThrottled() {
+			// Update RetryAfterReader so that no more requests would be sent until RetryAfter expires.
+			c.RetryAfterWriter = rerr.RetryAfter
+		}
+
+		return rerr
+	}
+	return nil
+}
+
+// createOrUpdateVirtualNetworkLink creates or updates a virtual network link.
+func (c *Client) createOrUpdateVirtualNetworkLink(ctx context.Context, resourceGroupName, privateZoneName, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, etag string, waitForCompletion bool) *retry.Error {
+	resourceID := armclient.GetChildResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		privateDNSZoneResourceType,
+		privateZoneName,
+		virtualNetworkLinkResourceType,
+		virtualNetworkLinkName,
+	)
+	decorators := []autorest.PrepareDecorator{
+		autorest.WithPathParameters("{resourceID}", map[string]interface{}{"resourceID": resourceID}),
+		autorest.WithJSON(parameters),
+	}
+	if etag != "" {
+		decorators = append(decorators, autorest.WithHeader("If-Match", autorest.String(etag)))
+	}
+
+	var response *http.Response
+	var rerr *retry.Error
 	if waitForCompletion {
-		err := createOrUpdateFuture.WaitForCompletionRef(ctx, c.virtualNetworkLinksClient.Client)
-		if err != nil {
-			klog.V(5).Infof("Received error while waiting for completion for %s, resourceGroup: %s, privateZoneName: %s, error: %s", "virtualnetworklinks.put.request", resourceGroupName, privateZoneName, err)
-			return err
+		response, rerr = c.armClient.PutResourceWithDecorators(ctx, resourceID, parameters, decorators)
+	} else {
+		response, rerr = c.armClient.PutResourceWithDecoratorsAsync(ctx, resourceID, parameters, decorators)
+	}
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "virtualnetworklink.put.request", resourceID, rerr.Error())
+		return rerr
+	}
+
+	if !waitForCompletion {
+		return nil
+	}
+
+	if response != nil && response.StatusCode != http.StatusNoContent {
+		_, rerr = c.createOrUpdateResponder(response)
+		if rerr != nil {
+			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "virtualnetworklink.put.request", resourceID, rerr.Error())
+			return rerr
 		}
 	}
 	return nil
 }
 
+func (c *Client) createOrUpdateResponder(resp *http.Response) (*privatedns.VirtualNetworkLink, *retry.Error) {
+	result := &privatedns.VirtualNetworkLink{}
+	err := autorest.Respond(
+		resp,
+		azure.WithErrorUnlessStatusCode(http.StatusOK, http.StatusCreated),
+		autorest.ByUnmarshallingJSON(&result))
+	result.Response = autorest.Response{Response: resp}
+	return result, retry.GetError(resp, err)
+}
+
 // Get gets a virtual network link
-func (c *Client) Get(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string) (result privatedns.VirtualNetworkLink, err error) {
-	return c.virtualNetworkLinksClient.Get(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName)
+func (c *Client) Get(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string) (result privatedns.VirtualNetworkLink, err *retry.Error) {
+	mc := metrics.NewMetricContext("virtual_network_links", "get", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterReader.TryAccept() {
+		mc.RateLimitedCount()
+		return privatedns.VirtualNetworkLink{}, retry.GetRateLimitError(false, "VirtualNetworkLinkGet")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterReader.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("VirtualNetworkLinkGet", "client throttled", c.RetryAfterReader)
+		return privatedns.VirtualNetworkLink{}, rerr
+	}
+	result, rerr := c.getVirtualNetworkLink(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName)
+
+	mc.Observe(rerr)
+	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
+}
+
+// getVirtualNetworkLink gets a virtual network link.
+func (c *Client) getVirtualNetworkLink(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string) (privatedns.VirtualNetworkLink, *retry.Error) {
+	resourceID := armclient.GetChildResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		privateDNSZoneResourceType,
+		privateZoneName,
+		virtualNetworkLinkResourceType,
+		virtualNetworkLinkName,
+	)
+	result := privatedns.VirtualNetworkLink{}
+	response, rerr := c.armClient.GetResourceWithExpandQuery(ctx, resourceID, "")
+	defer c.armClient.CloseResponse(ctx, response)
+	if rerr != nil {
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "virtualnetworklink.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", "virtualnetworklink.get.respond", resourceID, err)
+		return result, retry.GetError(response, err)
+	}
+
+	result.Response = autorest.Response{Response: response}
+	return result, nil
 }
diff --git a/pkg/azureclients/virtualnetworklinksclient/doc.go b/pkg/azureclients/virtualnetworklinksclient/doc.go
new file mode 100644
index 000000000..219f7df49
--- /dev/null
+++ b/pkg/azureclients/virtualnetworklinksclient/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2022 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 virtualnetworklinksclient implements the client for Virtual Network Link.
+package virtualnetworklinksclient // import "sigs.k8s.io/cloud-provider-azure/pkg/azureclients/virtualnetworklinksclient"
diff --git a/pkg/azureclients/virtualnetworklinksclient/interface.go b/pkg/azureclients/virtualnetworklinksclient/interface.go
index a4af65819..ff8c6c2df 100644
--- a/pkg/azureclients/virtualnetworklinksclient/interface.go
+++ b/pkg/azureclients/virtualnetworklinksclient/interface.go
@@ -20,15 +20,23 @@ import (
 	"context"
 
 	"github.com/Azure/azure-sdk-for-go/services/privatedns/mgmt/2018-09-01/privatedns"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+const (
+	// APIVersion is the API version.
+	// 2020-06-01 is the latest supported version
+	APIVersion = "2020-06-01"
+	// AzureStackCloudName is the cloud name of Azure Stack
+	AzureStackCloudName = "AZURESTACKCLOUD"
 )
 
 // Interface is the client interface for Virtual Network Link.
 // Don't forget to run "hack/update-mock-clients.sh" command to generate the mock client.
 type Interface interface {
-
 	// Get gets a virtual network link
-	Get(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string) (result privatedns.VirtualNetworkLink, err error)
+	Get(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string) (privatedns.VirtualNetworkLink, *retry.Error)
 
-	// CreateOrUpdate creates or updates a private dns zone.
-	CreateOrUpdate(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, waitForCompletion bool) error
+	// CreateOrUpdate creates or updates a virtual network link.
+	CreateOrUpdate(ctx context.Context, resourceGroupName string, privateZoneName string, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, etag string, waitForCompletion bool) *retry.Error
 }
diff --git a/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/doc.go b/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/doc.go
index 575cacf0e..23cc1c691 100644
--- a/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/doc.go
+++ b/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/doc.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Kubernetes Authors.
+Copyright 2022 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.
diff --git a/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/interface.go b/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/interface.go
index 105dd4dfd..22311f9e3 100644
--- a/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/interface.go
+++ b/pkg/azureclients/virtualnetworklinksclient/mockvirtualnetworklinksclient/interface.go
@@ -27,6 +27,7 @@ import (
 
 	privatedns "github.com/Azure/azure-sdk-for-go/services/privatedns/mgmt/2018-09-01/privatedns"
 	gomock "github.com/golang/mock/gomock"
+	retry "sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
 // MockInterface is a mock of Interface interface.
@@ -53,25 +54,25 @@ func (m *MockInterface) EXPECT() *MockInterfaceMockRecorder {
 }
 
 // CreateOrUpdate mocks base method.
-func (m *MockInterface) CreateOrUpdate(ctx context.Context, resourceGroupName, privateZoneName, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, waitForCompletion bool) error {
+func (m *MockInterface) CreateOrUpdate(ctx context.Context, resourceGroupName, privateZoneName, virtualNetworkLinkName string, parameters privatedns.VirtualNetworkLink, etag string, waitForCompletion bool) *retry.Error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateOrUpdate", ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, waitForCompletion)
-	ret0, _ := ret[0].(error)
+	ret := m.ctrl.Call(m, "CreateOrUpdate", ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, etag, waitForCompletion)
+	ret0, _ := ret[0].(*retry.Error)
 	return ret0
 }
 
 // CreateOrUpdate indicates an expected call of CreateOrUpdate.
-func (mr *MockInterfaceMockRecorder) CreateOrUpdate(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, waitForCompletion interface{}) *gomock.Call {
+func (mr *MockInterfaceMockRecorder) CreateOrUpdate(ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, etag, waitForCompletion interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateOrUpdate", reflect.TypeOf((*MockInterface)(nil).CreateOrUpdate), ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, waitForCompletion)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateOrUpdate", reflect.TypeOf((*MockInterface)(nil).CreateOrUpdate), ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName, parameters, etag, waitForCompletion)
 }
 
 // Get mocks base method.
-func (m *MockInterface) Get(ctx context.Context, resourceGroupName, privateZoneName, virtualNetworkLinkName string) (privatedns.VirtualNetworkLink, error) {
+func (m *MockInterface) Get(ctx context.Context, resourceGroupName, privateZoneName, virtualNetworkLinkName string) (privatedns.VirtualNetworkLink, *retry.Error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Get", ctx, resourceGroupName, privateZoneName, virtualNetworkLinkName)
 	ret0, _ := ret[0].(privatedns.VirtualNetworkLink)
-	ret1, _ := ret[1].(error)
+	ret1, _ := ret[1].(*retry.Error)
 	return ret0, ret1
 }
 
diff --git a/pkg/provider/azure_storageaccount.go b/pkg/provider/azure_storageaccount.go
index 8357b45ed..5cfc49413 100644
--- a/pkg/provider/azure_storageaccount.go
+++ b/pkg/provider/azure_storageaccount.go
@@ -308,7 +308,8 @@ func (az *Cloud) createPrivateEndpoint(ctx context.Context, accountName string,
 		Location:                  &az.Location,
 		PrivateEndpointProperties: &network.PrivateEndpointProperties{Subnet: &subnet, PrivateLinkServiceConnections: &privateLinkServiceConnections},
 	}
-	return az.privateendpointclient.CreateOrUpdate(ctx, vnetResourceGroup, privateEndpointName, privateEndpoint, true)
+
+	return az.privateendpointclient.CreateOrUpdate(ctx, vnetResourceGroup, privateEndpointName, privateEndpoint, "", true).Error()
 }
 
 func (az *Cloud) createPrivateDNSZone(ctx context.Context, vnetResourceGroup string) error {
@@ -335,7 +336,7 @@ func (az *Cloud) createVNetLink(ctx context.Context, vNetLinkName, vnetResourceG
 			VirtualNetwork:      &privatedns.SubResource{ID: &vnetID},
 			RegistrationEnabled: to.BoolPtr(true)},
 	}
-	return az.virtualNetworkLinksClient.CreateOrUpdate(ctx, vnetResourceGroup, PrivateDNSZoneName, vNetLinkName, parameters, false)
+	return az.virtualNetworkLinksClient.CreateOrUpdate(ctx, vnetResourceGroup, PrivateDNSZoneName, vNetLinkName, parameters, "", false).Error()
 }
 
 func (az *Cloud) createPrivateDNSZoneGroup(ctx context.Context, dnsZoneGroupName, privateEndpointName, vnetResourceGroup string) error {


Test patch:
diff --git a/pkg/azureclients/privateendpointclient/azure_privateendpointclient_test.go b/pkg/azureclients/privateendpointclient/azure_privateendpointclient_test.go
new file mode 100644
index 000000000..dc6d1fdfa
--- /dev/null
+++ b/pkg/azureclients/privateendpointclient/azure_privateendpointclient_test.go
@@ -0,0 +1,183 @@
+/*
+Copyright 2022 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 privatelinkserviceclient implements the client for PrivateLinkService.
+package privateendpointclient
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/to"
+	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/" + PEResourceType + "/pe1"
+	testResourcePrefix = "/subscriptions/subscriptionID/resourceGroups/rg/providers/" + PEResourceType
+)
+
+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},
+	}
+
+	peClient := New(config)
+	assert.Equal(t, "sub", peClient.subscriptionID)
+	assert.NotEmpty(t, peClient.rateLimiterReader)
+	assert.NotEmpty(t, peClient.rateLimiterWriter)
+}
+
+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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	peClient := getTestPrivateEndpointClient(armClient)
+	expected := network.PrivateEndpoint{Response: autorest.Response{}}
+	result, rerr := peClient.Get(context.TODO(), "rg", "pe1", "")
+	assert.Equal(t, expected, 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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	peClient := getTestPrivateEndpointClient(armClient)
+	expected := network.PrivateEndpoint{Response: autorest.Response{}}
+	result, rerr := peClient.Get(context.TODO(), "rg", "pe1", "")
+	assert.Equal(t, expected, 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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, throttleErr).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	peClient := getTestPrivateEndpointClient(armClient)
+	result, rerr := peClient.Get(context.TODO(), "rg", "pe1", "")
+	assert.Empty(t, result)
+	assert.Equal(t, throttleErr, rerr)
+}
+
+func TestCreateOrUpdate(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	pe := getTestPrivateEndpoint("pe1")
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte(""))),
+	}
+	armClient.EXPECT().PutResourceWithDecorators(gomock.Any(), to.String(pe.ID), pe, gomock.Any()).Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	peClient := getTestPrivateEndpointClient(armClient)
+	rerr := peClient.CreateOrUpdate(context.TODO(), "rg", "pe1", pe, "", true)
+	assert.Nil(t, rerr)
+}
+
+func TestCreateOrUpdateAsync(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	pe := getTestPrivateEndpoint("pe1")
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte(""))),
+	}
+	armClient.EXPECT().PutResourceWithDecoratorsAsync(gomock.Any(), to.String(pe.ID), pe, gomock.Any()).Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	peClient := getTestPrivateEndpointClient(armClient)
+	rerr := peClient.CreateOrUpdate(context.TODO(), "rg", "pe1", pe, "", false)
+	assert.Nil(t, rerr)
+}
+
+func getTestPrivateEndpoint(name string) network.PrivateEndpoint {
+	return network.PrivateEndpoint{
+		ID:       to.StringPtr(fmt.Sprintf("%s/%s", testResourcePrefix, name)),
+		Name:     to.StringPtr(name),
+		Location: to.StringPtr("eastus"),
+	}
+}
+
+func getTestPrivateEndpointClient(armClient armclient.Interface) *Client {
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(&azclients.RateLimitConfig{})
+	return &Client{
+		armClient:         armClient,
+		subscriptionID:    "subscriptionID",
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+	}
+}
diff --git a/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient_test.go b/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient_test.go
new file mode 100644
index 000000000..f41d6e0c8
--- /dev/null
+++ b/pkg/azureclients/virtualnetworklinksclient/azure_virtualnetworklinksclient_test.go
@@ -0,0 +1,183 @@
+/*
+Copyright 2022 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 virtualnetworklinksclient implements the client for VirtualNetworkLinks.
+package virtualnetworklinksclient
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/services/privatedns/mgmt/2018-09-01/privatedns"
+	"github.com/Azure/go-autorest/autorest"
+	"github.com/Azure/go-autorest/autorest/to"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	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/" + privateDNSZoneResourceType + "/pz1/" + virtualNetworkLinkResourceType + "/vnl1"
+	testResourceIDFormat = "/subscriptions/subscriptionID/resourceGroups/rg/providers/" + privateDNSZoneResourceType + "/%s/" + virtualNetworkLinkResourceType + "/%s"
+)
+
+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},
+	}
+
+	vnlClient := New(config)
+	assert.Equal(t, "sub", vnlClient.subscriptionID)
+	assert.NotEmpty(t, vnlClient.rateLimiterReader)
+	assert.NotEmpty(t, vnlClient.rateLimiterWriter)
+}
+
+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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vnlClient := getTestVirtualNetworkLinkClient(armClient)
+	expected := privatedns.VirtualNetworkLink{Response: autorest.Response{}}
+	result, rerr := vnlClient.Get(context.TODO(), "rg", "pz1", "vnl1")
+	assert.Equal(t, expected, 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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vnlClient := getTestVirtualNetworkLinkClient(armClient)
+	expected := privatedns.VirtualNetworkLink{Response: autorest.Response{}}
+	result, rerr := vnlClient.Get(context.TODO(), "rg", "pz1", "vnl1")
+	assert.Equal(t, expected, 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().GetResourceWithExpandQuery(gomock.Any(), testResourceID, "").Return(response, throttleErr).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vnlClient := getTestVirtualNetworkLinkClient(armClient)
+	result, rerr := vnlClient.Get(context.TODO(), "rg", "pz1", "vnl1")
+	assert.Empty(t, result)
+	assert.Equal(t, throttleErr, rerr)
+}
+
+func TestCreateOrUpdate(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vnl := getTestVirtualNetworkLink("pz1", "vnl1")
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte(""))),
+	}
+	armClient.EXPECT().PutResourceWithDecorators(gomock.Any(), to.String(vnl.ID), vnl, gomock.Any()).Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vnlClient := getTestVirtualNetworkLinkClient(armClient)
+	rerr := vnlClient.CreateOrUpdate(context.TODO(), "rg", "pz1", "vnl1", vnl, "", true)
+	assert.Nil(t, rerr)
+}
+
+func TestCreateOrUpdateAsync(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	vnl := getTestVirtualNetworkLink("pz1", "vnl1")
+	armClient := mockarmclient.NewMockInterface(ctrl)
+	response := &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte(""))),
+	}
+	armClient.EXPECT().PutResourceWithDecoratorsAsync(gomock.Any(), to.String(vnl.ID), vnl, gomock.Any()).Return(response, nil).Times(1)
+	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
+
+	vnlClient := getTestVirtualNetworkLinkClient(armClient)
+	rerr := vnlClient.CreateOrUpdate(context.TODO(), "rg", "pz1", "vnl1", vnl, "", false)
+	assert.Nil(t, rerr)
+}
+
+func getTestVirtualNetworkLink(privateZoneName, virtualNetworkLinkName string) privatedns.VirtualNetworkLink {
+	return privatedns.VirtualNetworkLink{
+		ID:       to.StringPtr(fmt.Sprintf(testResourceIDFormat, privateZoneName, virtualNetworkLinkName)),
+		Name:     to.StringPtr(virtualNetworkLinkName),
+		Location: to.StringPtr("eastus"),
+	}
+}
+
+func getTestVirtualNetworkLinkClient(armClient armclient.Interface) *Client {
+	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(&azclients.RateLimitConfig{})
+	return &Client{
+		armClient:         armClient,
+		subscriptionID:    "subscriptionID",
+		rateLimiterReader: rateLimiterReader,
+		rateLimiterWriter: rateLimiterWriter,
+	}
+}
diff --git a/pkg/provider/azure_storageaccount_test.go b/pkg/provider/azure_storageaccount_test.go
index ea51b3016..480aa262a 100644
--- a/pkg/provider/azure_storageaccount_test.go
+++ b/pkg/provider/azure_storageaccount_test.go
@@ -375,11 +375,11 @@ func TestEnsureStorageAccountWithPrivateEndpoint(t *testing.T) {
 	cloud.privatednszonegroupclient = mockPrivateDNSZoneGroup
 
 	mockPrivateEndpointClient := mockprivateendpointclient.NewMockInterface(ctrl)
-	mockPrivateEndpointClient.EXPECT().CreateOrUpdate(gomock.Any(), vnetResourceGroup, gomock.Any(), gomock.Any(), true).Return(nil).Times(1)
+	mockPrivateEndpointClient.EXPECT().CreateOrUpdate(gomock.Any(), vnetResourceGroup, gomock.Any(), gomock.Any(), "", true).Return(nil).Times(1)
 	cloud.privateendpointclient = mockPrivateEndpointClient
 
 	mockVirtualNetworkLinksClient := mockvirtualnetworklinksclient.NewMockInterface(ctrl)
-	mockVirtualNetworkLinksClient.EXPECT().CreateOrUpdate(gomock.Any(), vnetResourceGroup, gomock.Any(), gomock.Any(), gomock.Any(), false).Return(nil).Times(1)
+	mockVirtualNetworkLinksClient.EXPECT().CreateOrUpdate(gomock.Any(), vnetResourceGroup, gomock.Any(), gomock.Any(), gomock.Any(), "", false).Return(nil).Times(1)
 	cloud.virtualNetworkLinksClient = mockVirtualNetworkLinksClient
 
 	testAccountOptions := &AccountOptions{