Repository: kubernetes-sigs/cloud-provider-azure
Pull request #4391 (2023-07-31 09:09:30)

Feat: add load balancer backend pool batch updater
<!--  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
/kind bug

#### What this PR does / why we need it:

feat: add load balancer backend pool batch updater
fix a bug that the route updater and zone fetcher will not be exited every time the service reconciler restarts.

#### 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: https://github.com/kubernetes-sigs/cloud-provider-azure/issues/4013

#### 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
feat: add load balancer backend pool batch updater
```

#### 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/consts/consts.go b/pkg/consts/consts.go
index cbed3f4d0..a0b9c10df 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -535,3 +535,10 @@ const (
 const (
 	VMSSTagForBatchOperation = "aks-managed-coordination"
 )
+
+type LoadBalancerBackendPoolUpdateOperation string
+
+const (
+	LoadBalancerBackendPoolUpdateOperationAdd    LoadBalancerBackendPoolUpdateOperation = "add"
+	LoadBalancerBackendPoolUpdateOperationRemove LoadBalancerBackendPoolUpdateOperation = "remove"
+)
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index 4f5df51ad..0332e17e4 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -405,7 +405,7 @@ type Cloud struct {
 	KubeClient       clientset.Interface
 	eventBroadcaster record.EventBroadcaster
 	eventRecorder    record.EventRecorder
-	routeUpdater     *delayedRouteUpdater
+	routeUpdater     batchProcessor
 
 	vmCache  azcache.Resource
 	lbCache  azcache.Resource
@@ -711,7 +711,7 @@ func (az *Cloud) InitializeCloudFromConfig(ctx context.Context, config *Config,
 	if callFromCCM {
 		// start delayed route updater.
 		az.routeUpdater = newDelayedRouteUpdater(az, consts.RouteUpdateInterval)
-		go az.routeUpdater.run()
+		go az.routeUpdater.run(ctx)
 
 		// Azure Stack does not support zone at the moment
 		// https://docs.microsoft.com/en-us/azure-stack/user/azure-stack-network-differences?view=azs-2102
@@ -723,7 +723,7 @@ func (az *Cloud) InitializeCloudFromConfig(ctx context.Context, config *Config,
 				return err
 			}
 
-			go az.refreshZones(az.syncRegionZonesMap)
+			go az.refreshZones(ctx, az.syncRegionZonesMap)
 		}
 	}
 
diff --git a/pkg/provider/azure_loadbalancer_backendpool.go b/pkg/provider/azure_loadbalancer_backendpool.go
index 162832319..c1b195729 100644
--- a/pkg/provider/azure_loadbalancer_backendpool.go
+++ b/pkg/provider/azure_loadbalancer_backendpool.go
@@ -425,6 +425,7 @@ func (bi *backendPoolTypeNodeIP) EnsureHostsInPool(service *v1.Service, nodes []
 			}
 		}
 
+		var nodeIPsToBeAdded []string
 		nodePrivateIPsSet := sets.New[string]()
 		for _, node := range nodes {
 			if isControlPlaneNode(node) {
@@ -444,21 +445,12 @@ func (bi *backendPoolTypeNodeIP) EnsureHostsInPool(service *v1.Service, nodes []
 
 			if !existingIPs.Has(privateIP) {
 				name := node.Name
-				if utilnet.IsIPv6String(privateIP) {
-					name = fmt.Sprintf("%s-%s", name, v6Suffix)
-				}
-
 				klog.V(6).Infof("bi.EnsureHostsInPool: adding %s with ip address %s", name, privateIP)
-				*backendPool.LoadBalancerBackendAddresses = append(*backendPool.LoadBalancerBackendAddresses, network.LoadBalancerBackendAddress{
-					Name: pointer.String(name),
-					LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
-						IPAddress: pointer.String(privateIP),
-					},
-				})
+				nodeIPsToBeAdded = append(nodeIPsToBeAdded, privateIP)
 				numOfAdd++
-				changed = true
 			}
 		}
+		changed = bi.addNodeIPAddressesToBackendPool(&backendPool, nodeIPsToBeAdded)
 
 		var nodeIPsToBeDeleted []string
 		for _, loadBalancerBackendAddress := range *backendPool.LoadBalancerBackendAddresses {
@@ -772,6 +764,47 @@ func newBackendPool(lb *network.LoadBalancer, isBackendPoolPreConfigured bool, p
 	return isBackendPoolPreConfigured
 }
 
+func (az *Cloud) addNodeIPAddressesToBackendPool(backendPool *network.BackendAddressPool, nodeIPAddresses []string) bool {
+	if backendPool.LoadBalancerBackendAddresses == nil {
+		lbBackendPoolAddresses := make([]network.LoadBalancerBackendAddress, 0)
+		backendPool.LoadBalancerBackendAddresses = &lbBackendPoolAddresses
+	}
+
+	var changed bool
+	addresses := *backendPool.LoadBalancerBackendAddresses
+	for _, ipAddress := range nodeIPAddresses {
+		if !hasIPAddressInBackendPool(backendPool, ipAddress) {
+			name := az.nodePrivateIPToNodeNameMap[ipAddress]
+			klog.V(4).Infof("bi.addNodeIPAddressesToBackendPool: adding %s to the backend pool %s", ipAddress, pointer.StringDeref(backendPool.Name, ""))
+			addresses = append(addresses, network.LoadBalancerBackendAddress{
+				Name: pointer.String(name),
+				LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
+					IPAddress: pointer.String(ipAddress),
+				},
+			})
+			changed = true
+		}
+	}
+	backendPool.LoadBalancerBackendAddresses = &addresses
+	return changed
+}
+
+func hasIPAddressInBackendPool(backendPool *network.BackendAddressPool, ipAddress string) bool {
+	if backendPool.LoadBalancerBackendAddresses == nil {
+		return false
+	}
+
+	addresses := *backendPool.LoadBalancerBackendAddresses
+	for _, address := range addresses {
+		if address.LoadBalancerBackendAddressPropertiesFormat != nil &&
+			pointer.StringDeref(address.IPAddress, "") == ipAddress {
+			return true
+		}
+	}
+
+	return false
+}
+
 func removeNodeIPAddressesFromBackendPool(
 	backendPool network.BackendAddressPool,
 	nodeIPAddresses []string,
diff --git a/pkg/provider/azure_local_services.go b/pkg/provider/azure_local_services.go
new file mode 100644
index 000000000..1cee6717f
--- /dev/null
+++ b/pkg/provider/azure_local_services.go
@@ -0,0 +1,220 @@
+/*
+Copyright 2023 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 provider
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+// batchProcessor collects operations in a certain interval and then processes them in batches.
+type batchProcessor interface {
+	// run starts the batchProcessor, and stops if the context exits.
+	run(ctx context.Context)
+
+	// addOperation adds an operation to the batchProcessor.
+	addOperation(operation batchOperation) batchOperation
+}
+
+// batchOperation is an operation that can be added to a batchProcessor.
+type batchOperation interface {
+	wait() batchOperationResult
+}
+
+// loadBalancerBackendPoolUpdateOperation is an operation that updates the backend pool of a load balancer.
+type loadBalancerBackendPoolUpdateOperation struct {
+	loadBalancerName string
+	backendPoolName  string
+	kind             consts.LoadBalancerBackendPoolUpdateOperation
+	nodeIPs          []string
+	result           chan batchOperationResult
+}
+
+func (op *loadBalancerBackendPoolUpdateOperation) wait() batchOperationResult {
+	return <-op.result
+}
+
+// loadBalancerBackendPoolUpdater is a batchProcessor that updates the backend pool of a load balancer.
+type loadBalancerBackendPoolUpdater struct {
+	az         *Cloud
+	interval   time.Duration
+	lock       sync.Mutex
+	operations []batchOperation
+}
+
+// newLoadBalancerBackendPoolUpdater creates a new loadBalancerBackendPoolUpdater.
+func newLoadBalancerBackendPoolUpdater(az *Cloud, interval time.Duration) *loadBalancerBackendPoolUpdater {
+	return &loadBalancerBackendPoolUpdater{
+		az:         az,
+		interval:   interval,
+		operations: make([]batchOperation, 0),
+	}
+}
+
+// run starts the loadBalancerBackendPoolUpdater, and stops if the context exits.
+func (updater *loadBalancerBackendPoolUpdater) run(ctx context.Context) {
+	err := wait.PollUntilContextCancel(ctx, updater.interval, false, func(ctx context.Context) (bool, error) {
+		updater.process()
+		return false, nil
+	})
+	if err != nil {
+		klog.Infof("loadBalancerBackendPoolUpdater.run: stopped due to %s", err.Error())
+	}
+}
+
+// getAddIPsToBackendPoolOperation creates a new loadBalancerBackendPoolUpdateOperation
+// that adds nodeIPs to the backend pool.
+func getAddIPsToBackendPoolOperation(loadBalancerName, backendPoolName string, nodeIPs []string) *loadBalancerBackendPoolUpdateOperation {
+	return &loadBalancerBackendPoolUpdateOperation{
+		loadBalancerName: loadBalancerName,
+		backendPoolName:  backendPoolName,
+		kind:             consts.LoadBalancerBackendPoolUpdateOperationAdd,
+		nodeIPs:          nodeIPs,
+		result:           make(chan batchOperationResult),
+	}
+}
+
+// getRemoveIPsFromBackendPoolOperation creates a new loadBalancerBackendPoolUpdateOperation
+// that removes nodeIPs from the backend pool.
+func getRemoveIPsFromBackendPoolOperation(loadBalancerName, backendPoolName string, nodeIPs []string) *loadBalancerBackendPoolUpdateOperation {
+	return &loadBalancerBackendPoolUpdateOperation{
+		loadBalancerName: loadBalancerName,
+		backendPoolName:  backendPoolName,
+		kind:             consts.LoadBalancerBackendPoolUpdateOperationRemove,
+		nodeIPs:          nodeIPs,
+		result:           make(chan batchOperationResult),
+	}
+}
+
+// addOperation adds an operation to the loadBalancerBackendPoolUpdater.
+func (updater *loadBalancerBackendPoolUpdater) addOperation(operation batchOperation) batchOperation {
+	updater.lock.Lock()
+	defer updater.lock.Unlock()
+
+	updater.operations = append(updater.operations, operation)
+	return operation
+}
+
+// process processes all operations in the loadBalancerBackendPoolUpdater.
+// It merges operations that have the same loadBalancerName and backendPoolName,
+// and then processes them in batches. If an operation fails, it will be retried
+// if it is retriable, otherwise all operations in the batch targeting to
+// this backend pool will fail.
+func (updater *loadBalancerBackendPoolUpdater) process() {
+	updater.lock.Lock()
+	defer updater.lock.Unlock()
+
+	if len(updater.operations) == 0 {
+		klog.V(4).Infof("loadBalancerBackendPoolUpdater.process: no operations to process")
+		return
+	}
+
+	// Group operations by loadBalancerName:backendPoolName
+	groups := make(map[string][]batchOperation)
+	for _, op := range updater.operations {
+		lbOp := op.(*loadBalancerBackendPoolUpdateOperation)
+		key := fmt.Sprintf("%s:%s", lbOp.loadBalancerName, lbOp.backendPoolName)
+		groups[key] = append(groups[key], op)
+	}
+
+	// Clear all jobs.
+	updater.operations = make([]batchOperation, 0)
+
+	for key, ops := range groups {
+		parts := strings.Split(key, ":")
+		lbName, poolName := parts[0], parts[1]
+		operationName := fmt.Sprintf("%s/%s", lbName, poolName)
+		bp, rerr := updater.az.LoadBalancerClient.GetLBBackendPool(context.Background(), updater.az.ResourceGroup, lbName, poolName, "")
+		if rerr != nil {
+			updater.processError(rerr, operationName, ops...)
+			continue
+		}
+
+		var changed bool
+		for _, op := range ops {
+			lbOp := op.(*loadBalancerBackendPoolUpdateOperation)
+			switch lbOp.kind {
+			case consts.LoadBalancerBackendPoolUpdateOperationRemove:
+				changed = removeNodeIPAddressesFromBackendPool(bp, lbOp.nodeIPs, false, true)
+			case consts.LoadBalancerBackendPoolUpdateOperationAdd:
+				changed = updater.az.addNodeIPAddressesToBackendPool(&bp, lbOp.nodeIPs)
+			default:
+				panic("loadBalancerBackendPoolUpdater.process: unknown operation type")
+			}
+		}
+		// To keep the code clean, ignore the case when `changed` is true
+		// but the backend pool object is not changed after multiple times of removal and re-adding.
+		if changed {
+			rerr = updater.az.LoadBalancerClient.CreateOrUpdateBackendPools(context.Background(), updater.az.ResourceGroup, lbName, poolName, bp, pointer.StringDeref(bp.Etag, ""))
+			if rerr != nil {
+				updater.processError(rerr, operationName, ops...)
+				continue
+			}
+		}
+		notify(newBatchOperationResult(operationName, true, nil), ops...)
+	}
+}
+
+// processError mark the operations as retriable if the error is retriable,
+// and fail all operations if the error is not retriable.
+func (updater *loadBalancerBackendPoolUpdater) processError(
+	rerr *retry.Error,
+	operationName string,
+	operations ...batchOperation,
+) {
+	if rerr.Retriable {
+		// Retry if retriable.
+		updater.operations = append(updater.operations, operations...)
+	} else {
+		// Fail all operations if not retriable.
+		notify(newBatchOperationResult(operationName, false, rerr.Error()), operations...)
+	}
+}
+
+// notify notifies the operations with the result.
+func notify(res batchOperationResult, operations ...batchOperation) {
+	for _, op := range operations {
+		lbOp := op.(*loadBalancerBackendPoolUpdateOperation)
+		lbOp.result <- res
+	}
+}
+
+// batchOperationResult is the result of a batch operation.
+type batchOperationResult struct {
+	name    string
+	success bool
+	err     error
+}
+
+// newBatchOperationResult creates a new batchOperationResult.
+func newBatchOperationResult(name string, success bool, err error) batchOperationResult {
+	return batchOperationResult{
+		name:    name,
+		success: success,
+		err:     err,
+	}
+}
diff --git a/pkg/provider/azure_routes.go b/pkg/provider/azure_routes.go
index ecb6ca642..d78fb3f8b 100644
--- a/pkg/provider/azure_routes.go
+++ b/pkg/provider/azure_routes.go
@@ -54,11 +54,11 @@ type delayedRouteOperation struct {
 	route          network.Route
 	routeTableTags map[string]*string
 	operation      routeOperation
-	result         chan error
+	result         chan batchOperationResult
 }
 
 // wait waits for the operation completion and returns the result.
-func (op *delayedRouteOperation) wait() error {
+func (op *delayedRouteOperation) wait() batchOperationResult {
 	return <-op.result
 }
 
@@ -72,27 +72,26 @@ type delayedRouteUpdater struct {
 	interval time.Duration
 
 	lock           sync.Mutex
-	routesToUpdate []*delayedRouteOperation
+	routesToUpdate []batchOperation
 }
 
 // newDelayedRouteUpdater creates a new delayedRouteUpdater.
-func newDelayedRouteUpdater(az *Cloud, interval time.Duration) *delayedRouteUpdater {
+func newDelayedRouteUpdater(az *Cloud, interval time.Duration) batchProcessor {
 	return &delayedRouteUpdater{
 		az:             az,
 		interval:       interval,
-		routesToUpdate: make([]*delayedRouteOperation, 0),
+		routesToUpdate: make([]batchOperation, 0),
 	}
 }
 
 // run starts the updater reconciling loop.
-func (d *delayedRouteUpdater) run() {
-	err := wait.PollImmediateInfinite(d.interval, func() (bool, error) {
+func (d *delayedRouteUpdater) run(ctx context.Context) {
+	klog.Info("delayedRouteUpdater: started")
+	err := wait.PollUntilContextCancel(ctx, d.interval, true, func(ctx context.Context) (bool, error) {
 		d.updateRoutes()
 		return false, nil
 	})
-	if err != nil { // this should never happen, if it does, panic
-		panic(err)
-	}
+	klog.Info("delayedRouteUpdater: stopped due to %s", err.Error())
 }
 
 // updateRoutes invokes route table client to update all routes.
@@ -109,11 +108,12 @@ func (d *delayedRouteUpdater) updateRoutes() {
 	var err error
 	defer func() {
 		// Notify all the goroutines.
-		for _, rt := range d.routesToUpdate {
-			rt.result <- err
+		for _, op := range d.routesToUpdate {
+			rt := op.(*delayedRouteOperation)
+			rt.result <- newBatchOperationResult("", false, err)
 		}
 		// Clear all the jobs.
-		d.routesToUpdate = make([]*delayedRouteOperation, 0)
+		d.routesToUpdate = make([]batchOperation, 0)
 	}()
 
 	var (
@@ -153,7 +153,8 @@ func (d *delayedRouteUpdater) updateRoutes() {
 		onlyUpdateTags = false
 	}
 
-	for _, rt := range d.routesToUpdate {
+	for _, op := range d.routesToUpdate {
+		rt := op.(*delayedRouteOperation)
 		if rt.operation == routeTableOperationUpdateTags {
 			routeTable.Tags = rt.routeTableTags
 			dirty = true
@@ -238,32 +239,37 @@ func (d *delayedRouteUpdater) cleanupOutdatedRoutes(existingRoutes []network.Rou
 	return existingRoutes, changed
 }
 
-// addRouteOperation adds the routeOperation to delayedRouteUpdater and returns a delayedRouteOperation.
-func (d *delayedRouteUpdater) addRouteOperation(operation routeOperation, route network.Route) (*delayedRouteOperation, error) {
-	d.lock.Lock()
-	defer d.lock.Unlock()
+func getAddRouteOperation(route network.Route) batchOperation {
+	return &delayedRouteOperation{
+		route:     route,
+		operation: routeOperationAdd,
+		result:    make(chan batchOperationResult),
+	}
+}
 
-	op := &delayedRouteOperation{
+func getDeleteRouteOperation(route network.Route) batchOperation {
+	return &delayedRouteOperation{
 		route:     route,
-		operation: operation,
-		result:    make(chan error),
+		operation: routeOperationDelete,
+		result:    make(chan batchOperationResult),
+	}
+}
+
+func getUpdateRouteTableTagsOperation(tags map[string]*string) batchOperation {
+	return &delayedRouteOperation{
+		routeTableTags: tags,
+		operation:      routeTableOperationUpdateTags,
+		result:         make(chan batchOperationResult),
 	}
-	d.routesToUpdate = append(d.routesToUpdate, op)
-	return op, nil
 }
 
-// addUpdateRouteTableTagsOperation adds a update route table tags operation to delayedRouteUpdater and returns a delayedRouteOperation.
-func (d *delayedRouteUpdater) addUpdateRouteTableTagsOperation(operation routeOperation, tags map[string]*string) (*delayedRouteOperation, error) {
+// addOperation adds the routeOperation to delayedRouteUpdater and returns a delayedRouteOperation.
+func (d *delayedRouteUpdater) addOperation(operation batchOperation) batchOperation {
 	d.lock.Lock()
 	defer d.lock.Unlock()
 
-	op := &delayedRouteOperation{
-		routeTableTags: tags,
-		operation:      operation,
-		result:         make(chan error),
-	}
-	d.routesToUpdate = append(d.routesToUpdate, op)
-	return op, nil
+	d.routesToUpdate = append(d.routesToUpdate, operation)
+	return operation
 }
 
 // ListRoutes lists all managed routes that belong to the specified clusterName
@@ -296,14 +302,10 @@ func (az *Cloud) ListRoutes(ctx context.Context, clusterName string) ([]*cloudpr
 	tags, changed := az.ensureRouteTableTagged(&routeTable)
 	if changed {
 		klog.V(2).Infof("ListRoutes: updating tags on route table %s", pointer.StringDeref(routeTable.Name, ""))
-		op, err := az.routeUpdater.addUpdateRouteTableTagsOperation(routeTableOperationUpdateTags, tags)
-		if err != nil {
-			klog.Errorf("ListRoutes: failed to add route table operation with error: %v", err)
-			return nil, err
-		}
+		op := az.routeUpdater.addOperation(getUpdateRouteTableTagsOperation(tags))
 
 		// Wait for operation complete.
-		err = op.wait()
+		err = op.wait().err
 		if err != nil {
 			klog.Errorf("ListRoutes: failed to update route table tags with error: %v", err)
 			return nil, err
@@ -421,14 +423,10 @@ func (az *Cloud) CreateRoute(ctx context.Context, clusterName string, nameHint s
 	}
 
 	klog.V(2).Infof("CreateRoute: creating route for clusterName=%q instance=%q cidr=%q", clusterName, kubeRoute.TargetNode, kubeRoute.DestinationCIDR)
-	op, err := az.routeUpdater.addRouteOperation(routeOperationAdd, route)
-	if err != nil {
-		klog.Errorf("CreateRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
-		return err
-	}
+	op := az.routeUpdater.addOperation(getAddRouteOperation(route))
 
 	// Wait for operation complete.
-	err = op.wait()
+	err = op.wait().err
 	if err != nil {
 		klog.Errorf("CreateRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
 		return err
@@ -469,14 +467,10 @@ func (az *Cloud) DeleteRoute(ctx context.Context, clusterName string, kubeRoute
 		Name:                  pointer.String(routeName),
 		RoutePropertiesFormat: &network.RoutePropertiesFormat{},
 	}
-	op, err := az.routeUpdater.addRouteOperation(routeOperationDelete, route)
-	if err != nil {
-		klog.Errorf("DeleteRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
-		return err
-	}
+	op := az.routeUpdater.addOperation(getDeleteRouteOperation(route))
 
 	// Wait for operation complete.
-	err = op.wait()
+	err = op.wait().err
 	if err != nil {
 		klog.Errorf("DeleteRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
 		return err
@@ -490,14 +484,10 @@ func (az *Cloud) DeleteRoute(ctx context.Context, clusterName string, kubeRoute
 			Name:                  pointer.String(routeNameWithoutIPV6Suffix),
 			RoutePropertiesFormat: &network.RoutePropertiesFormat{},
 		}
-		op, err := az.routeUpdater.addRouteOperation(routeOperationDelete, route)
-		if err != nil {
-			klog.Errorf("DeleteRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
-			return err
-		}
+		op := az.routeUpdater.addOperation(getDeleteRouteOperation(route))
 
 		// Wait for operation complete.
-		err = op.wait()
+		err = op.wait().err
 		if err != nil {
 			klog.Errorf("DeleteRoute failed for node %q with error: %v", kubeRoute.TargetNode, err)
 			return err
diff --git a/pkg/provider/azure_zones.go b/pkg/provider/azure_zones.go
index 68f8f8203..7c922ac84 100644
--- a/pkg/provider/azure_zones.go
+++ b/pkg/provider/azure_zones.go
@@ -23,7 +23,6 @@ import (
 	"os"
 	"strconv"
 	"strings"
-	"time"
 
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -35,13 +34,13 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
 )
 
-func (az *Cloud) refreshZones(refreshFunc func() error) {
-	ticker := time.NewTicker(consts.ZoneFetchingInterval)
-	defer ticker.Stop()
-
-	for range ticker.C {
+func (az *Cloud) refreshZones(ctx context.Context, refreshFunc func() error) {
+	klog.V(2).Info("refreshZones: refreshing zones every 30 minutes.")
+	err := wait.PollUntilContextCancel(ctx, consts.ZoneFetchingInterval, false, func(ctx context.Context) (bool, error) {
 		_ = refreshFunc()
-	}
+		return false, nil
+	})
+	klog.V(2).Infof("refreshZones: refresh zones finished with error: %s", err.Error())
 }
 
 func (az *Cloud) syncRegionZonesMap() error {


Test patch:
diff --git a/pkg/provider/azure_loadbalancer_backendpool_test.go b/pkg/provider/azure_loadbalancer_backendpool_test.go
index e165a9ce9..177274332 100644
--- a/pkg/provider/azure_loadbalancer_backendpool_test.go
+++ b/pkg/provider/azure_loadbalancer_backendpool_test.go
@@ -44,6 +44,14 @@ func TestEnsureHostsInPoolNodeIP(t *testing.T) {
 
 	az := GetTestCloud(ctrl)
 	az.LoadBalancerSku = consts.LoadBalancerSkuStandard
+	az.nodePrivateIPToNodeNameMap = map[string]string{
+		"10.0.0.2": "vmss-0",
+		"2001::2":  "vmss-0",
+		"10.0.0.1": "vmss-1",
+		"2001::1":  "vmss-1",
+		"10.0.0.4": "vmss-2",
+		"2001::4":  "vmss-2",
+	}
 	bi := newBackendPoolTypeNodeIP(az)
 
 	nodes := []*v1.Node{
@@ -182,13 +190,13 @@ func TestEnsureHostsInPoolNodeIP(t *testing.T) {
 							},
 						},
 						{
-							Name: pointer.String("vmss-0-IPv6"),
+							Name: pointer.String("vmss-0"),
 							LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
 								IPAddress: pointer.String("2001::2"),
 							},
 						},
 						{
-							Name: pointer.String("vmss-2-IPv6"),
+							Name: pointer.String("vmss-2"),
 							LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
 								IPAddress: pointer.String("2001::4"),
 							},
diff --git a/pkg/provider/azure_local_services_test.go b/pkg/provider/azure_local_services_test.go
new file mode 100644
index 000000000..6cdaba35f
--- /dev/null
+++ b/pkg/provider/azure_local_services_test.go
@@ -0,0 +1,437 @@
+/*
+Copyright 2023 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 provider
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2022-07-01/network"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/utils/pointer"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/loadbalancerclient/mockloadbalancerclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+func TestLoadBalancerBackendPoolUpdater(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	addOperationPool1 := getAddIPsToBackendPoolOperation("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"})
+	removeOperationPool1 := getRemoveIPsFromBackendPoolOperation("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"})
+	addOperationPool2 := getAddIPsToBackendPoolOperation("lb1", "pool2", []string{"10.0.0.1", "10.0.0.2"})
+
+	testCases := []struct {
+		name                               string
+		operations                         []batchOperation
+		existingBackendPools               []network.BackendAddressPool
+		expectedGetBackendPool             network.BackendAddressPool
+		extraWait                          bool
+		expectedCreateOrUpdateBackendPools []network.BackendAddressPool
+		expectedBackendPools               []network.BackendAddressPool
+		expectedResult                     map[batchOperationResult]bool
+	}{
+		{
+			name:       "Add node IPs to backend pool",
+			operations: []batchOperation{addOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+			},
+		},
+		{
+			name:       "Remove node IPs from backend pool",
+			operations: []batchOperation{addOperationPool1, removeOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+			},
+		},
+		{
+			name:       "Multiple operations targeting different backend pools",
+			operations: []batchOperation{addOperationPool1, addOperationPool2, removeOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+				getTestBackendAddressPoolWithIPs("lb1", "pool2", []string{}),
+			},
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+				getTestBackendAddressPoolWithIPs("lb1", "pool2", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+				getTestBackendAddressPoolWithIPs("lb1", "pool2", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+				newBatchOperationResult("lb1/pool2", true, nil): true,
+			},
+		},
+		{
+			name:       "Multiple operations in two batches",
+			operations: []batchOperation{addOperationPool1, removeOperationPool1},
+			extraWait:  true,
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedGetBackendPool: getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cloud := GetTestCloud(ctrl)
+			mockLBClient := mockloadbalancerclient.NewMockInterface(ctrl)
+			mockLBClient.EXPECT().GetLBBackendPool(
+				gomock.Any(),
+				gomock.Any(),
+				"lb1",
+				*tc.existingBackendPools[0].Name,
+				gomock.Any(),
+			).Return(tc.existingBackendPools[0], nil)
+			if len(tc.existingBackendPools) == 2 {
+				mockLBClient.EXPECT().GetLBBackendPool(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.existingBackendPools[1].Name,
+					gomock.Any(),
+				).Return(tc.existingBackendPools[1], nil)
+			}
+			if tc.extraWait {
+				mockLBClient.EXPECT().GetLBBackendPool(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.expectedGetBackendPool.Name,
+					gomock.Any(),
+				).Return(tc.expectedGetBackendPool, nil)
+			}
+			mockLBClient.EXPECT().CreateOrUpdateBackendPools(
+				gomock.Any(),
+				gomock.Any(),
+				"lb1",
+				*tc.expectedCreateOrUpdateBackendPools[0].Name,
+				tc.expectedCreateOrUpdateBackendPools[0],
+				gomock.Any(),
+			).Return(nil)
+			if len(tc.existingBackendPools) == 2 || tc.extraWait {
+				mockLBClient.EXPECT().CreateOrUpdateBackendPools(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.expectedCreateOrUpdateBackendPools[1].Name,
+					tc.expectedCreateOrUpdateBackendPools[1],
+					gomock.Any(),
+				).Return(nil)
+			}
+			cloud.LoadBalancerClient = mockLBClient
+
+			u := newLoadBalancerBackendPoolUpdater(cloud, time.Second)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go u.run(ctx)
+
+			results := sync.Map{}
+			for _, op := range tc.operations {
+				op := op
+				go func() {
+					u.addOperation(op)
+					result := op.wait()
+					results.Store(result, true)
+				}()
+				time.Sleep(100 * time.Millisecond)
+				if tc.extraWait {
+					time.Sleep(time.Second)
+				}
+			}
+			err := wait.PollUntilContextTimeout(context.Background(), time.Second, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+				resultLen := 0
+				results.Range(func(key, value interface{}) bool {
+					resultLen++
+					return true
+				})
+				return resultLen == len(tc.expectedResult), nil
+			})
+			assert.NoError(t, err)
+
+			actualResults := make(map[batchOperationResult]bool)
+			results.Range(func(key, value interface{}) bool {
+				actualResults[key.(batchOperationResult)] = true
+				return true
+			})
+
+			err = wait.PollUntilContextTimeout(ctx, time.Second, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+				return assert.Equal(t, tc.expectedResult, actualResults, tc.name), nil
+			})
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func TestLoadBalancerBackendPoolUpdaterFailed(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	addOperationPool1 := getAddIPsToBackendPoolOperation("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"})
+
+	testCases := []struct {
+		name                               string
+		operations                         []batchOperation
+		existingBackendPools               []network.BackendAddressPool
+		expectedGetBackendPool             network.BackendAddressPool
+		getBackendPoolErr                  *retry.Error
+		putBackendPoolErr                  *retry.Error
+		expectedCreateOrUpdateBackendPools []network.BackendAddressPool
+		expectedBackendPools               []network.BackendAddressPool
+		expectedResult                     map[batchOperationResult]bool
+		expectedPoolNameToErrMsg           map[string]string
+		expectedResultCount                int
+	}{
+		{
+			name:       "Retriable error when getting backend pool",
+			operations: []batchOperation{addOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			getBackendPoolErr: retry.NewError(true, errors.New("error")),
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+			},
+			expectedResultCount: 1,
+		},
+		{
+			name:       "Retriable error when updating backend pool",
+			operations: []batchOperation{addOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedGetBackendPool: getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			putBackendPoolErr:      retry.NewError(true, errors.New("error")),
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedResult: map[batchOperationResult]bool{
+				newBatchOperationResult("lb1/pool1", true, nil): true,
+			},
+			expectedResultCount: 1,
+		},
+		{
+			name:       "Non-retriable error when getting backend pool",
+			operations: []batchOperation{addOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			getBackendPoolErr: retry.NewError(false, fmt.Errorf("error")),
+			expectedBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedPoolNameToErrMsg: map[string]string{"lb1/pool1": "Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"},
+			expectedResultCount:      1,
+		},
+		{
+			name:       "Non-retriable error when updating backend pool",
+			operations: []batchOperation{addOperationPool1},
+			existingBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			},
+			expectedGetBackendPool: getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{}),
+			putBackendPoolErr:      retry.NewError(false, fmt.Errorf("error")),
+			expectedCreateOrUpdateBackendPools: []network.BackendAddressPool{
+				getTestBackendAddressPoolWithIPs("lb1", "pool1", []string{"10.0.0.1", "10.0.0.2"}),
+			},
+			expectedPoolNameToErrMsg: map[string]string{"lb1/pool1": "Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"},
+			expectedResultCount:      1,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cloud := GetTestCloud(ctrl)
+			mockLBClient := mockloadbalancerclient.NewMockInterface(ctrl)
+			mockLBClient.EXPECT().GetLBBackendPool(
+				gomock.Any(),
+				gomock.Any(),
+				"lb1",
+				*tc.existingBackendPools[0].Name,
+				gomock.Any(),
+			).Return(tc.existingBackendPools[0], tc.getBackendPoolErr)
+			if tc.getBackendPoolErr != nil && tc.getBackendPoolErr.Retriable {
+				mockLBClient.EXPECT().GetLBBackendPool(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.existingBackendPools[0].Name,
+					gomock.Any(),
+				).Return(tc.existingBackendPools[0], nil)
+			}
+			if len(tc.existingBackendPools) == 2 {
+				mockLBClient.EXPECT().GetLBBackendPool(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.existingBackendPools[1].Name,
+					gomock.Any(),
+				).Return(tc.existingBackendPools[1], nil)
+			}
+			if tc.putBackendPoolErr != nil && tc.putBackendPoolErr.Retriable {
+				mockLBClient.EXPECT().GetLBBackendPool(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.expectedGetBackendPool.Name,
+					gomock.Any(),
+				).Return(tc.expectedGetBackendPool, nil)
+			}
+			if len(tc.expectedCreateOrUpdateBackendPools) > 0 {
+				mockLBClient.EXPECT().CreateOrUpdateBackendPools(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.expectedCreateOrUpdateBackendPools[0].Name,
+					tc.expectedCreateOrUpdateBackendPools[0],
+					gomock.Any(),
+				).Return(tc.putBackendPoolErr)
+			}
+			if len(tc.expectedCreateOrUpdateBackendPools) == 2 {
+				mockLBClient.EXPECT().CreateOrUpdateBackendPools(
+					gomock.Any(),
+					gomock.Any(),
+					"lb1",
+					*tc.expectedCreateOrUpdateBackendPools[1].Name,
+					tc.expectedCreateOrUpdateBackendPools[1],
+					gomock.Any(),
+				).Return(nil)
+			}
+			cloud.LoadBalancerClient = mockLBClient
+
+			u := newLoadBalancerBackendPoolUpdater(cloud, time.Second)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go u.run(ctx)
+
+			results := sync.Map{}
+			for _, op := range tc.operations {
+				op := op
+				go func() {
+					u.addOperation(op)
+					result := op.wait()
+					results.Store(result, true)
+				}()
+				time.Sleep(100 * time.Millisecond)
+			}
+			err := wait.PollUntilContextTimeout(context.Background(), time.Second, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+				resultLen := 0
+				results.Range(func(key, value interface{}) bool {
+					resultLen++
+					return true
+				})
+				return resultLen == tc.expectedResultCount, nil
+			})
+			assert.NoError(t, err)
+
+			actualResults := make(map[batchOperationResult]bool)
+			poolNameToErrMsg := make(map[string]string)
+			results.Range(func(key, value interface{}) bool {
+				actualResults[key.(batchOperationResult)] = true
+				if tc.getBackendPoolErr != nil && !tc.getBackendPoolErr.Retriable ||
+					tc.putBackendPoolErr != nil && !tc.putBackendPoolErr.Retriable {
+					poolNameToErrMsg[key.(batchOperationResult).name] = key.(batchOperationResult).err.Error()
+				}
+				return true
+			})
+
+			err = wait.PollUntilContextTimeout(ctx, time.Second, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+				if tc.expectedResult != nil {
+					return assert.Equal(t, tc.expectedResult, actualResults, tc.name), nil
+				}
+				if tc.expectedPoolNameToErrMsg != nil {
+					return assert.Equal(t, tc.expectedPoolNameToErrMsg, poolNameToErrMsg, tc.name), nil
+				}
+				return false, errors.New("unexpected result")
+			})
+			assert.NoError(t, err)
+		})
+	}
+}
+
+func getTestBackendAddressPoolWithIPs(lbName, bpName string, ips []string) network.BackendAddressPool {
+	bp := network.BackendAddressPool{
+		ID:   pointer.String(fmt.Sprintf("/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/%s/backendAddressPools/%s", lbName, bpName)),
+		Name: pointer.String(bpName),
+		BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+			Location:                     pointer.String("eastus"),
+			LoadBalancerBackendAddresses: &[]network.LoadBalancerBackendAddress{},
+		},
+	}
+	for _, ip := range ips {
+		if len(ip) > 0 {
+			*bp.LoadBalancerBackendAddresses = append(*bp.LoadBalancerBackendAddresses, network.LoadBalancerBackendAddress{
+				Name: pointer.String(""),
+				LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
+					IPAddress: pointer.String(ip),
+				},
+			})
+		}
+	}
+	return bp
+}
diff --git a/pkg/provider/azure_routes_test.go b/pkg/provider/azure_routes_test.go
index 23502f92d..ca49b9ec4 100644
--- a/pkg/provider/azure_routes_test.go
+++ b/pkg/provider/azure_routes_test.go
@@ -55,7 +55,7 @@ func TestDeleteRoute(t *testing.T) {
 	cache, _ := cloud.newRouteTableCache()
 	cloud.rtCache = cache
 	cloud.routeUpdater = newDelayedRouteUpdater(cloud, 100*time.Millisecond)
-	go cloud.routeUpdater.run()
+	go cloud.routeUpdater.run(context.Background())
 	route := cloudprovider.Route{
 		TargetNode:      "node",
 		DestinationCIDR: "1.2.3.4/24",
@@ -128,7 +128,7 @@ func TestDeleteRouteDualStack(t *testing.T) {
 	cache, _ := cloud.newRouteTableCache()
 	cloud.rtCache = cache
 	cloud.routeUpdater = newDelayedRouteUpdater(cloud, 100*time.Millisecond)
-	go cloud.routeUpdater.run()
+	go cloud.routeUpdater.run(context.Background())
 
 	route := cloudprovider.Route{
 		TargetNode:      "node",
@@ -200,7 +200,7 @@ func TestCreateRoute(t *testing.T) {
 	cache, _ := cloud.newRouteTableCache()
 	cloud.rtCache = cache
 	cloud.routeUpdater = newDelayedRouteUpdater(cloud, 100*time.Millisecond)
-	go cloud.routeUpdater.run()
+	go cloud.routeUpdater.run(context.Background())
 
 	route := cloudprovider.Route{TargetNode: "node", DestinationCIDR: "1.2.3.4/24"}
 	nodePrivateIP := "2.4.6.8"
@@ -607,7 +607,7 @@ func TestListRoutes(t *testing.T) {
 	cache, _ := cloud.newRouteTableCache()
 	cloud.rtCache = cache
 	cloud.routeUpdater = newDelayedRouteUpdater(cloud, 100*time.Millisecond)
-	go cloud.routeUpdater.run()
+	go cloud.routeUpdater.run(context.Background())
 
 	testCases := []struct {
 		name                  string
diff --git a/pkg/provider/azure_zones_test.go b/pkg/provider/azure_zones_test.go
index 0161b5def..99c421072 100644
--- a/pkg/provider/azure_zones_test.go
+++ b/pkg/provider/azure_zones_test.go
@@ -23,6 +23,7 @@ import (
 	"net"
 	"net/http"
 	"testing"
+	"time"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/zoneclient/mockzoneclient"
 
@@ -375,3 +376,20 @@ func TestGetRegionZonesBackoff(t *testing.T) {
 		})
 	}
 }
+
+func TestRefreshZones(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	az := GetTestCloud(ctrl)
+	mockZoneClient := mockzoneclient.NewMockInterface(ctrl)
+	mockZoneClient.EXPECT().GetZones(gomock.Any(), gomock.Any()).Return(map[string][]string{}, nil).Times(0)
+	az.ZoneClient = mockZoneClient
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		time.Sleep(1 * time.Second)
+		cancel()
+	}()
+	az.refreshZones(ctx, az.syncRegionZonesMap)
+}