Repository: kubernetes-sigs/cloud-provider-azure
Pull request #966 (2022-01-04 05:12:44)

feat: support sending sync requests concurrently in batches when putt…
…ing vmss vm

(cherry picked from commit d9edcd6ec1429d8995d985b1ebddbce77bfd25ad)

<!--  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?

<!--
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
-->

/kind feature

#### What this PR does / why we need it:

Cherry pick of #953: feat: support sending sync requests concurrently in batches when putting vmss vm

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
Introduce a configuration option `putVMSSVMBatchSize`. If set, the sync requests will be sent concurrently in batches when putting vmss vms.

```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/966#" 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/release-1.0/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 or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-capz-1-21 | f94a7652c43e407b95cce816e0d0b9c1d5ef7d02 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/966/pull-cloud-provider-azure-e2e-ccm-capz-1-21/1478232911843954688) | false | `/test pull-cloud-provider-azure-e2e-ccm-capz-1-21`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=966). [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 or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-e2e-ccm-capz-1-21 | f94a7652c43e407b95cce816e0d0b9c1d5ef7d02 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/966/pull-cloud-provider-azure-e2e-ccm-capz-1-21/1478258918906925056) | false | `/test pull-cloud-provider-azure-e2e-ccm-capz-1-21`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=966). [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/armclient/azure_armclient.go b/pkg/azureclients/armclient/azure_armclient.go
index 72d4b8942..90f2a34d4 100644
--- a/pkg/azureclients/armclient/azure_armclient.go
+++ b/pkg/azureclients/armclient/azure_armclient.go
@@ -361,46 +361,7 @@ func (c *Client) PutResource(ctx context.Context, resourceID string, parameters
 	return c.PutResourceWithDecorators(ctx, resourceID, parameters, putDecorators)
 }
 
-// PutResources puts a list of resources from resources map[resourceID]parameters.
-// Those resources sync requests are sequential while async requests are concurrent. It's especially
-// useful when the ARM API doesn't support concurrent requests.
-func (c *Client) PutResources(ctx context.Context, resources map[string]interface{}) map[string]*PutResourcesResponse {
-	if len(resources) == 0 {
-		return nil
-	}
-
-	// Sequential sync requests.
-	futures := make(map[string]*azure.Future)
-	responses := make(map[string]*PutResourcesResponse)
-	for resourceID, parameters := range resources {
-		decorators := []autorest.PrepareDecorator{
-			autorest.WithPathParameters("{resourceID}", map[string]interface{}{"resourceID": resourceID}),
-			autorest.WithJSON(parameters),
-		}
-		request, err := c.PreparePutRequest(ctx, decorators...)
-		if err != nil {
-			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "put.prepare", resourceID, err)
-			responses[resourceID] = &PutResourcesResponse{
-				Error: retry.NewError(false, err),
-			}
-			continue
-		}
-		dumpRequest(request, 10)
-
-		future, resp, clientErr := c.SendAsync(ctx, request)
-		defer c.CloseResponse(ctx, resp)
-		if clientErr != nil {
-			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "put.send", resourceID, clientErr.Error())
-			responses[resourceID] = &PutResourcesResponse{
-				Error: clientErr,
-			}
-			continue
-		}
-
-		futures[resourceID] = future
-	}
-
-	// Concurrent async requests.
+func (c *Client) waitAsync(ctx context.Context, futures map[string]*azure.Future, previousResponses map[string]*PutResourcesResponse) {
 	wg := sync.WaitGroup{}
 	var responseLock sync.Mutex
 	for resourceID, future := range futures {
@@ -423,22 +384,111 @@ func (c *Client) PutResources(ctx context.Context, resources map[string]interfac
 				}
 
 				responseLock.Lock()
-				responses[resourceID] = &PutResourcesResponse{
+				previousResponses[resourceID] = &PutResourcesResponse{
 					Error: retriableErr,
 				}
 				responseLock.Unlock()
 				return
 			}
+		}(resourceID, future)
+	}
+	wg.Wait()
+}
+
+// PutResources puts a list of resources from resources map[resourceID]parameters.
+// Those resources sync requests are sequential while async requests are concurrent. It's especially
+// useful when the ARM API doesn't support concurrent requests.
+func (c *Client) PutResources(ctx context.Context, resources map[string]interface{}) map[string]*PutResourcesResponse {
+	if len(resources) == 0 {
+		return nil
+	}
 
-			responseLock.Lock()
+	// Sequential sync requests.
+	futures := make(map[string]*azure.Future)
+	responses := make(map[string]*PutResourcesResponse)
+	for resourceID, parameters := range resources {
+		future, rerr := c.PutResourceAsync(ctx, resourceID, parameters)
+		if rerr != nil {
 			responses[resourceID] = &PutResourcesResponse{
-				Response: response,
+				Error: rerr,
 			}
-			responseLock.Unlock()
-		}(resourceID, future)
+			continue
+		}
+		futures[resourceID] = future
 	}
 
-	wg.Wait()
+	c.waitAsync(ctx, futures, responses)
+
+	return responses
+}
+
+// PutResourcesInBatches is similar with PutResources, but it sends sync request concurrently in batches.
+func (c *Client) PutResourcesInBatches(ctx context.Context, resources map[string]interface{}, batchSize int) map[string]*PutResourcesResponse {
+	if len(resources) == 0 {
+		return nil
+	}
+
+	if batchSize <= 0 {
+		klog.V(4).Infof("PutResourcesInBatches: batch size %d, put resources in sequence", batchSize)
+		return c.PutResources(ctx, resources)
+	}
+
+	if batchSize > len(resources) {
+		klog.V(4).Infof("PutResourcesInBatches: batch size %d, but the number of the resources is %d", batchSize, resources)
+		batchSize = len(resources)
+	}
+	klog.V(4).Infof("PutResourcesInBatches: send sync requests in parallel with the batch size %d", batchSize)
+
+	// Convert map to slice because it is more straightforward to
+	// loop over slice in batches than map.
+	type resourcesMeta struct {
+		resourceID string
+		parameters interface{}
+	}
+	resourcesList := make([]resourcesMeta, 0)
+	for resourceID, parameters := range resources {
+		resourcesList = append(resourcesList, resourcesMeta{
+			resourceID: resourceID,
+			parameters: parameters,
+		})
+	}
+
+	// Concurrent sync requests in batches.
+	futures := make(map[string]*azure.Future)
+	responses := make(map[string]*PutResourcesResponse)
+	wg := sync.WaitGroup{}
+	var responseLock, futuresLock sync.Mutex
+	for i := 0; i < len(resourcesList); i += batchSize {
+		j := i + batchSize
+		if j > len(resourcesList) {
+			j = len(resourcesList)
+		}
+
+		for k := i; k < j; k++ {
+			wg.Add(1)
+			go func(resourceID string, parameters interface{}) {
+				defer wg.Done()
+				future, rerr := c.PutResourceAsync(ctx, resourceID, parameters)
+				if rerr != nil {
+					responseLock.Lock()
+					responses[resourceID] = &PutResourcesResponse{
+						Error: rerr,
+					}
+					responseLock.Unlock()
+					return
+				}
+
+				futuresLock.Lock()
+				futures[resourceID] = future
+				futuresLock.Unlock()
+			}(resourcesList[k].resourceID, resourcesList[k].parameters)
+		}
+		wg.Wait()
+	}
+
+	// Concurrent async requests.
+	c.waitAsync(ctx, futures, responses)
+
 	return responses
 }
 
diff --git a/pkg/azureclients/armclient/interface.go b/pkg/azureclients/armclient/interface.go
index e7d437e3e..0926c903b 100644
--- a/pkg/azureclients/armclient/interface.go
+++ b/pkg/azureclients/armclient/interface.go
@@ -71,6 +71,9 @@ type Interface interface {
 	// useful when the ARM API doesn't support concurrent requests.
 	PutResources(ctx context.Context, resources map[string]interface{}) map[string]*PutResourcesResponse
 
+	// PutResourcesInBatches is similar with PutResources, but it sends sync request concurrently in batches.
+	PutResourcesInBatches(ctx context.Context, resources map[string]interface{}, batchSize int) map[string]*PutResourcesResponse
+
 	// PutResourceWithDecorators puts a resource with decorators by resource ID
 	PutResourceWithDecorators(ctx context.Context, resourceID string, parameters interface{}, decorators []autorest.PrepareDecorator) (*http.Response, *retry.Error)
 
diff --git a/pkg/azureclients/armclient/mockarmclient/interface.go b/pkg/azureclients/armclient/mockarmclient/interface.go
index bd1246839..5802fa8a7 100644
--- a/pkg/azureclients/armclient/mockarmclient/interface.go
+++ b/pkg/azureclients/armclient/mockarmclient/interface.go
@@ -240,6 +240,20 @@ func (mr *MockInterfaceMockRecorder) PutResources(ctx, resources interface{}) *g
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PutResources", reflect.TypeOf((*MockInterface)(nil).PutResources), ctx, resources)
 }
 
+// PutResourcesInBatches mocks base method
+func (m *MockInterface) PutResourcesInBatches(ctx context.Context, resources map[string]interface{}, batchSize int) map[string]*armclient.PutResourcesResponse {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PutResourcesInBatches", ctx, resources, batchSize)
+	ret0, _ := ret[0].(map[string]*armclient.PutResourcesResponse)
+	return ret0
+}
+
+// PutResourcesInBatches indicates an expected call of PutResourcesInBatches
+func (mr *MockInterfaceMockRecorder) PutResourcesInBatches(ctx, resources, batchSize interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PutResourcesInBatches", reflect.TypeOf((*MockInterface)(nil).PutResourcesInBatches), ctx, resources, batchSize)
+}
+
 // PutResourceWithDecorators mocks base method
 func (m *MockInterface) PutResourceWithDecorators(ctx context.Context, resourceID string, parameters interface{}, decorators []autorest.PrepareDecorator) (*http.Response, *retry.Error) {
 	m.ctrl.T.Helper()
diff --git a/pkg/azureclients/vmssvmclient/azure_vmssvmclient.go b/pkg/azureclients/vmssvmclient/azure_vmssvmclient.go
index a8716c6af..8205f336b 100644
--- a/pkg/azureclients/vmssvmclient/azure_vmssvmclient.go
+++ b/pkg/azureclients/vmssvmclient/azure_vmssvmclient.go
@@ -384,7 +384,9 @@ func (page VirtualMachineScaleSetVMListResultPage) Values() []compute.VirtualMac
 }
 
 // UpdateVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM.
-func (c *Client) UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string) *retry.Error {
+// If the batch size > 0, it will send sync requests concurrently in batches, or it will send sync requests in sequence.
+// No matter what the batch size is, it will process the async requests concurrently in one single batch.
+func (c *Client) UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string, batchSize int) *retry.Error {
 	mc := metrics.NewMetricContext("vmssvm", "update_vms", resourceGroupName, c.subscriptionID, source)
 
 	// Report errors if the client is rate limited.
@@ -400,7 +402,7 @@ func (c *Client) UpdateVMs(ctx context.Context, resourceGroupName string, VMScal
 		return rerr
 	}
 
-	rerr := c.updateVMSSVMs(ctx, resourceGroupName, VMScaleSetName, instances)
+	rerr := c.updateVMSSVMs(ctx, resourceGroupName, VMScaleSetName, instances, batchSize)
 	_ = mc.Observe(rerr.Error())
 	if rerr != nil {
 		if rerr.IsThrottled() {
@@ -415,7 +417,7 @@ func (c *Client) UpdateVMs(ctx context.Context, resourceGroupName string, VMScal
 }
 
 // updateVMSSVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM.
-func (c *Client) updateVMSSVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM) *retry.Error {
+func (c *Client) updateVMSSVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, batchSize int) *retry.Error {
 	resources := make(map[string]interface{})
 	for instanceID, parameter := range instances {
 		resourceID := armclient.GetChildResourceID(
@@ -429,7 +431,7 @@ func (c *Client) updateVMSSVMs(ctx context.Context, resourceGroupName string, VM
 		resources[resourceID] = parameter
 	}
 
-	responses := c.armClient.PutResources(ctx, resources)
+	responses := c.armClient.PutResourcesInBatches(ctx, resources, batchSize)
 	errors := make([]*retry.Error, 0)
 	for resourceID, resp := range responses {
 		if resp == nil {
diff --git a/pkg/azureclients/vmssvmclient/interface.go b/pkg/azureclients/vmssvmclient/interface.go
index 165e33215..9743a9a69 100644
--- a/pkg/azureclients/vmssvmclient/interface.go
+++ b/pkg/azureclients/vmssvmclient/interface.go
@@ -47,5 +47,5 @@ type Interface interface {
 	Update(ctx context.Context, resourceGroupName string, VMScaleSetName string, instanceID string, parameters compute.VirtualMachineScaleSetVM, source string) *retry.Error
 
 	// UpdateVMs updates a list of VirtualMachineScaleSetVM from map[instanceID]compute.VirtualMachineScaleSetVM.
-	UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string) *retry.Error
+	UpdateVMs(ctx context.Context, resourceGroupName string, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string, batchSize int) *retry.Error
 }
diff --git a/pkg/azureclients/vmssvmclient/mockvmssvmclient/interface.go b/pkg/azureclients/vmssvmclient/mockvmssvmclient/interface.go
index 30d240d30..cbbc21aff 100644
--- a/pkg/azureclients/vmssvmclient/mockvmssvmclient/interface.go
+++ b/pkg/azureclients/vmssvmclient/mockvmssvmclient/interface.go
@@ -93,15 +93,15 @@ func (mr *MockInterfaceMockRecorder) Update(ctx, resourceGroupName, VMScaleSetNa
 }
 
 // UpdateVMs mocks base method
-func (m *MockInterface) UpdateVMs(ctx context.Context, resourceGroupName, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string) *retry.Error {
+func (m *MockInterface) UpdateVMs(ctx context.Context, resourceGroupName, VMScaleSetName string, instances map[string]compute.VirtualMachineScaleSetVM, source string, batchSize int) *retry.Error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UpdateVMs", ctx, resourceGroupName, VMScaleSetName, instances, source)
+	ret := m.ctrl.Call(m, "UpdateVMs", ctx, resourceGroupName, VMScaleSetName, instances, source, batchSize)
 	ret0, _ := ret[0].(*retry.Error)
 	return ret0
 }
 
 // UpdateVMs indicates an expected call of UpdateVMs
-func (mr *MockInterfaceMockRecorder) UpdateVMs(ctx, resourceGroupName, VMScaleSetName, instances, source interface{}) *gomock.Call {
+func (mr *MockInterfaceMockRecorder) UpdateVMs(ctx, resourceGroupName, VMScaleSetName, instances, source, batchSize interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVMs", reflect.TypeOf((*MockInterface)(nil).UpdateVMs), ctx, resourceGroupName, VMScaleSetName, instances, source)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVMs", reflect.TypeOf((*MockInterface)(nil).UpdateVMs), ctx, resourceGroupName, VMScaleSetName, instances, source, batchSize)
 }
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index d598c94fc..443314f6c 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -225,6 +225,9 @@ type Config struct {
 	// 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"`
+	// PutVMSSVMBatchSize defines how many requests the client send concurrently when putting the VMSS VMs.
+	// If it is smaller than or equal to zero, the request will be sent one by one in sequence (default).
+	PutVMSSVMBatchSize int `json:"putVMSSVMBatchSize" yaml:"putVMSSVMBatchSize"`
 }
 
 type InitSecretConfig struct {
@@ -570,6 +573,10 @@ func (az *Cloud) InitializeCloudFromConfig(config *Config, fromSecret, syncZones
 	return nil
 }
 
+func (az *Cloud) getPutVMSSVMBatchSize() int {
+	return az.PutVMSSVMBatchSize
+}
+
 func (az *Cloud) initCaches() (err error) {
 	az.vmCache, err = az.newVMCache()
 	if err != nil {
diff --git a/pkg/provider/azure_vmss.go b/pkg/provider/azure_vmss.go
index 68224ba4a..9fe42198c 100644
--- a/pkg/provider/azure_vmss.go
+++ b/pkg/provider/azure_vmss.go
@@ -1319,7 +1319,7 @@ func (ss *ScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, bac
 			ctx, cancel := getContextWithCancel()
 			defer cancel()
 			klog.V(2).Infof("EnsureHostInPool begins to UpdateVMs for VMSS(%s, %s) with new backendPoolID %s", meta.resourceGroup, meta.vmssName, backendPoolID)
-			rerr := ss.VirtualMachineScaleSetVMsClient.UpdateVMs(ctx, meta.resourceGroup, meta.vmssName, update, "network_update")
+			rerr := ss.VirtualMachineScaleSetVMsClient.UpdateVMs(ctx, meta.resourceGroup, meta.vmssName, update, "network_update", ss.getPutVMSSVMBatchSize())
 			if rerr != nil {
 				klog.Errorf("EnsureHostInPool UpdateVMs for VMSS(%s, %s) failed with error %v", meta.resourceGroup, meta.vmssName, rerr.Error())
 				return rerr.Error()
@@ -1581,7 +1581,7 @@ func (ss *ScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoolID,
 			ctx, cancel := getContextWithCancel()
 			defer cancel()
 			klog.V(2).Infof("EnsureBackendPoolDeleted begins to UpdateVMs for VMSS(%s, %s) with backendPoolID %s", meta.resourceGroup, meta.vmssName, backendPoolID)
-			rerr := ss.VirtualMachineScaleSetVMsClient.UpdateVMs(ctx, meta.resourceGroup, meta.vmssName, update, "network_update")
+			rerr := ss.VirtualMachineScaleSetVMsClient.UpdateVMs(ctx, meta.resourceGroup, meta.vmssName, update, "network_update", ss.getPutVMSSVMBatchSize())
 			if rerr != nil {
 				klog.Errorf("EnsureBackendPoolDeleted UpdateVMs for VMSS(%s, %s) failed with error %v", meta.resourceGroup, meta.vmssName, rerr.Error())
 				return rerr.Error()


Test patch:
diff --git a/pkg/azureclients/armclient/azure_armclient_test.go b/pkg/azureclients/armclient/azure_armclient_test.go
index c9d80bc0c..2fc0dfd08 100644
--- a/pkg/azureclients/armclient/azure_armclient_test.go
+++ b/pkg/azureclients/armclient/azure_armclient_test.go
@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 	"time"
 
@@ -340,6 +341,26 @@ func TestPutResource(t *testing.T) {
 }
 
 func TestPutResources(t *testing.T) {
+	total := 0
+	server := getTestServer(t, &total)
+
+	backoff := &retry.Backoff{Steps: 1}
+	armClient := New(nil, server.URL, "test", "2019-01-01", "eastus", backoff)
+	armClient.client.RetryDuration = time.Millisecond * 1
+
+	ctx := context.Background()
+	resources := map[string]interface{}{
+		"/id/1": nil,
+		"/id/2": nil,
+	}
+	responses := armClient.PutResources(ctx, nil)
+	assert.Nil(t, responses)
+	responses = armClient.PutResources(ctx, resources)
+	assert.NotNil(t, responses)
+	assert.Equal(t, 3, total)
+}
+
+func getTestServer(t *testing.T, counter *int) *httptest.Server {
 	serverFuncs := []func(rw http.ResponseWriter, req *http.Request){
 		func(rw http.ResponseWriter, req *http.Request) {
 			assert.Equal(t, "PUT", req.Method)
@@ -369,30 +390,81 @@ func TestPutResources(t *testing.T) {
 		},
 	}
 
-	i, total := 0, 0
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	i := 0
+	var l sync.Mutex
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		l.Lock()
 		serverFuncs[i](w, r)
 		i++
 		if i > 3 {
 			i = 3
 		}
-		total++
+		*counter++
+		l.Unlock()
 	}))
+}
 
-	backoff := &retry.Backoff{Steps: 1}
-	armClient := New(nil, server.URL, "test", "2019-01-01", "eastus", backoff)
-	armClient.client.RetryDuration = time.Millisecond * 1
-
-	ctx := context.Background()
-	resources := map[string]interface{}{
-		"/id/1": nil,
-		"/id/2": nil,
+func TestPutResourcesInBatches(t *testing.T) {
+	for _, testCase := range []struct {
+		description                  string
+		resources                    map[string]interface{}
+		batchSize, expectedCallTimes int
+	}{
+		{
+			description: "",
+			resources: map[string]interface{}{
+				"/id/1": nil,
+				"/id/2": nil,
+			},
+			batchSize:         2,
+			expectedCallTimes: 3,
+		},
+		{
+			description: "",
+			resources: map[string]interface{}{
+				"/id/1": nil,
+				"/id/2": nil,
+			},
+			batchSize:         1,
+			expectedCallTimes: 3,
+		},
+		{
+			description: "",
+			resources:   nil,
+		},
+		{
+			description: "PutResourcesInBatches should set the batch size to the length of the resources if the batch size is larger than it",
+			resources: map[string]interface{}{
+				"/id/1": nil,
+				"/id/2": nil,
+			},
+			batchSize:         10,
+			expectedCallTimes: 3,
+		},
+		{
+			description: "PutResourcesInBatches should call PutResources if the batch size is smaller than or equal to zero",
+			resources: map[string]interface{}{
+				"/id/1": nil,
+				"/id/2": nil,
+			},
+			expectedCallTimes: 3,
+		},
+	} {
+		t.Run(testCase.description, func(t *testing.T) {
+			total := 0
+			server := getTestServer(t, &total)
+
+			backoff := &retry.Backoff{Steps: 1}
+			armClient := New(nil, server.URL, "test", "2019-01-01", "eastus", backoff)
+			armClient.client.RetryDuration = time.Millisecond * 1
+
+			ctx := context.Background()
+			responses := armClient.PutResourcesInBatches(ctx, testCase.resources, testCase.batchSize)
+			assert.Equal(t, testCase.resources == nil, responses == nil)
+			assert.Equal(t, testCase.expectedCallTimes, total)
+		})
 	}
-	responses := armClient.PutResources(ctx, nil)
-	assert.Nil(t, responses)
-	responses = armClient.PutResources(ctx, resources)
-	assert.NotNil(t, responses)
-	assert.Equal(t, 3, total)
+
 }
 
 func TestPutResourceAsync(t *testing.T) {
diff --git a/pkg/azureclients/vmssvmclient/azure_vmssvmclient_test.go b/pkg/azureclients/vmssvmclient/azure_vmssvmclient_test.go
index c1eb44132..d59e912db 100644
--- a/pkg/azureclients/vmssvmclient/azure_vmssvmclient_test.go
+++ b/pkg/azureclients/vmssvmclient/azure_vmssvmclient_test.go
@@ -619,11 +619,11 @@ func TestUpdateVMs(t *testing.T) {
 	}
 
 	armClient := mockarmclient.NewMockInterface(ctrl)
-	armClient.EXPECT().PutResources(gomock.Any(), testvmssVMs).Return(responses).Times(1)
+	armClient.EXPECT().PutResourcesInBatches(gomock.Any(), testvmssVMs, 0).Return(responses).Times(1)
 	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(2)
 
 	vmssvmClient := getTestVMSSVMClient(armClient)
-	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test")
+	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test", 0)
 	assert.Nil(t, rerr)
 }
 
@@ -648,11 +648,11 @@ func TestUpdateVMsWithUpdateVMsResponderError(t *testing.T) {
 		},
 	}
 	armClient := mockarmclient.NewMockInterface(ctrl)
-	armClient.EXPECT().PutResources(gomock.Any(), testvmssVMs).Return(responses).Times(1)
+	armClient.EXPECT().PutResourcesInBatches(gomock.Any(), testvmssVMs, 0).Return(responses).Times(1)
 	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
 
 	vmssvmClient := getTestVMSSVMClient(armClient)
-	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test")
+	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test", 0)
 	assert.NotNil(t, rerr)
 }
 
@@ -668,7 +668,7 @@ func TestUpdateVMsNeverRateLimiter(t *testing.T) {
 
 	armClient := mockarmclient.NewMockInterface(ctrl)
 	vmssvmClient := getTestVMSSVMClientWithNeverRateLimiter(armClient)
-	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test")
+	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test", 0)
 	assert.NotNil(t, rerr)
 	assert.Equal(t, vmssvmUpdateVMsErr, rerr)
 }
@@ -686,7 +686,7 @@ func TestUpdateVMsRetryAfterReader(t *testing.T) {
 
 	armClient := mockarmclient.NewMockInterface(ctrl)
 	vmClient := getTestVMSSVMClientWithRetryAfterReader(armClient)
-	rerr := vmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test")
+	rerr := vmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test", 0)
 	assert.NotNil(t, rerr)
 	assert.Equal(t, vmssvmUpdateVMsErr, rerr)
 }
@@ -719,11 +719,11 @@ func TestUpdateVMsThrottle(t *testing.T) {
 	}
 
 	armClient := mockarmclient.NewMockInterface(ctrl)
-	armClient.EXPECT().PutResources(gomock.Any(), testvmssVMs).Return(responses).Times(1)
+	armClient.EXPECT().PutResourcesInBatches(gomock.Any(), testvmssVMs, 0).Return(responses).Times(1)
 	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any()).Times(1)
 
 	vmssvmClient := getTestVMSSVMClient(armClient)
-	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test")
+	rerr := vmssvmClient.UpdateVMs(context.TODO(), "rg", "vmss1", instances, "test", 0)
 	assert.NotNil(t, rerr)
 	assert.EqualError(t, throttleErr.Error(), rerr.RawError.Error())
 }
diff --git a/pkg/provider/azure_vmss_test.go b/pkg/provider/azure_vmss_test.go
index aa1ae8ce7..3fc0284b7 100644
--- a/pkg/provider/azure_vmss_test.go
+++ b/pkg/provider/azure_vmss_test.go
@@ -2190,7 +2190,7 @@ func TestEnsureHostsInPool(t *testing.T) {
 		expectedVMSSVMs, _, _ := buildTestVirtualMachineEnv(ss.cloud, testVMSSName, "", 0, []string{"vmss-vm-000000", "vmss-vm-000001", "vmss-vm-000002"}, "", false)
 		mockVMSSVMClient := ss.cloud.VirtualMachineScaleSetVMsClient.(*mockvmssvmclient.MockInterface)
 		mockVMSSVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any()).Return(expectedVMSSVMs, nil).AnyTimes()
-		mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any(), gomock.Any()).Return(nil).Times(test.expectedVMSSVMPutTimes)
+		mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(test.expectedVMSSVMPutTimes)
 
 		mockVMClient := ss.cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
 		mockVMClient.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
@@ -2481,7 +2481,7 @@ func TestEnsureBackendPoolDeleted(t *testing.T) {
 		expectedVMSSVMs, _, _ := buildTestVirtualMachineEnv(ss.cloud, testVMSSName, "", 0, []string{"vmss-vm-000000", "vmss-vm-000001", "vmss-vm-000002"}, "", false)
 		mockVMSSVMClient := ss.cloud.VirtualMachineScaleSetVMsClient.(*mockvmssvmclient.MockInterface)
 		mockVMSSVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any()).Return(expectedVMSSVMs, nil).AnyTimes()
-		mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any(), gomock.Any()).Return(test.vmClientErr).Times(test.expectedVMSSVMPutTimes)
+		mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, testVMSSName, gomock.Any(), gomock.Any(), gomock.Any()).Return(test.vmClientErr).Times(test.expectedVMSSVMPutTimes)
 
 		err = ss.EnsureBackendPoolDeleted(&v1.Service{}, test.backendpoolID, testVMSSName, test.backendAddressPools, true)
 		assert.Equal(t, test.expectedErr, err != nil, test.description+", but an error occurs")
@@ -2554,7 +2554,7 @@ func TestEnsureBackendPoolDeletedConcurrently(t *testing.T) {
 	mockVMSSVMClient.EXPECT().List(gomock.Any(), "rg1", "vmss-0", gomock.Any()).Return(nil, nil).AnyTimes()
 	mockVMSSVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup, "vmss-0", gomock.Any()).Return(expectedVMSSVMsOfVMSS0, nil).AnyTimes()
 	mockVMSSVMClient.EXPECT().List(gomock.Any(), ss.ResourceGroup, "vmss-1", gomock.Any()).Return(expectedVMSSVMsOfVMSS1, nil).AnyTimes()
-	mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(2)
+	mockVMSSVMClient.EXPECT().UpdateVMs(gomock.Any(), ss.ResourceGroup, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(2)
 
 	backendpoolAddressIDs := []string{testLBBackendpoolID0, testLBBackendpoolID1, testLBBackendpoolID2}
 	testVMSSNames := []string{"vmss-0", "vmss-1", "vmss-2"}