Repository: kubernetes-sigs/cloud-provider-azure
Pull request #4033 (2023-06-02 07:40:42)

track2 sdk:add clientFactory and parse config from file
<!--  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 feature

<!--
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:
add clientFactory and parse config from file
add ratelimit policy
add throttled policy

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

#### Special notes for your reviewer:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
NONE
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/4033#" title="Author self-approved">MartinForReal</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fcloud-provider-azure).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 61089a440e864fa5fa433c90aac7b5cbb191e8f0 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/kubernetes-sigs-cloud-provide-azure/deploys/64799cfc47c39000080a2906 |
MartinForReal: /hold

Patch:
diff --git a/examples/azure.json b/examples/azure.json
index 44c672bf5..71ceb10ab 100644
--- a/examples/azure.json
+++ b/examples/azure.json
@@ -28,5 +28,6 @@
     "loadBalancerSku": "<loadbalancer-sku>",
     "excludeMasterFromStandardLB": false,
     "maximumLoadBalancerRuleCount": 250,
+    "enableMultipleStandardLoadBalancers": false,
     "tags": "a=b,c=d"
 }
\ No newline at end of file
diff --git a/examples/out-of-tree/cloud-controller-manager.yaml b/examples/out-of-tree/cloud-controller-manager.yaml
index 12b422efa..de8e25a76 100644
--- a/examples/out-of-tree/cloud-controller-manager.yaml
+++ b/examples/out-of-tree/cloud-controller-manager.yaml
@@ -252,5 +252,6 @@ stringData:
       "loadBalancerSku": "<loadbalancer-sku>",
       "excludeMasterFromStandardLB": false,
       "maximumLoadBalancerRuleCount": 250,
+      "enableMultipleStandardLoadBalancers": false,
       "tags": "a=b,c=d"
     }
diff --git a/hack/deploy-cluster-capz.sh b/hack/deploy-cluster-capz.sh
index d2b4ad3c8..f013b630d 100755
--- a/hack/deploy-cluster-capz.sh
+++ b/hack/deploy-cluster-capz.sh
@@ -45,6 +45,7 @@ export EXP_MACHINE_POOL=true
 export EXP_CLUSTER_RESOURCE_SET=true
 
 export AZURE_LOADBALANCER_SKU="${AZURE_LOADBALANCER_SKU:-Standard}"
+export ENABLE_MULTI_SLB="${ENABLE_MULTI_SLB:-false}"
 export LB_BACKEND_POOL_CONFIG_TYPE="${LB_BACKEND_POOL_CONFIG_TYPE:-nodeIPConfiguration}"
 export PUT_VMSS_VM_BATCH_SIZE="${PUT_VMSS_VM_BATCH_SIZE:-0}"
 
diff --git a/pkg/azclient/auth.go b/pkg/azclient/auth.go
new file mode 100644
index 000000000..b37fc6235
--- /dev/null
+++ b/pkg/azclient/auth.go
@@ -0,0 +1,252 @@
+/*
+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 azclient
+
+import (
+	"crypto/rsa"
+	"crypto/x509"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/arm/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
+	"golang.org/x/crypto/pkcs12"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/utils"
+)
+
+// AzureAuthConfig holds auth related part of cloud config
+type AzureAuthConfig struct {
+	// The AAD Tenant ID for the Subscription that the cluster is deployed in
+	TenantID string `json:"tenantId,omitempty" yaml:"tenantId,omitempty"`
+	// The ClientID for an AAD application with RBAC access to talk to Azure RM APIs
+	AADClientID string `json:"aadClientId,omitempty" yaml:"aadClientId,omitempty"`
+	// The ClientSecret for an AAD application with RBAC access to talk to Azure RM APIs
+	AADClientSecret string `json:"aadClientSecret,omitempty" yaml:"aadClientSecret,omitempty" datapolicy:"token"`
+	// The path of a client certificate for an AAD application with RBAC access to talk to Azure RM APIs
+	AADClientCertPath string `json:"aadClientCertPath,omitempty" yaml:"aadClientCertPath,omitempty"`
+	// The password of the client certificate for an AAD application with RBAC access to talk to Azure RM APIs
+	AADClientCertPassword string `json:"aadClientCertPassword,omitempty" yaml:"aadClientCertPassword,omitempty" datapolicy:"password"`
+	// Use managed service identity for the virtual machine to access Azure ARM APIs
+	UseManagedIdentityExtension bool `json:"useManagedIdentityExtension,omitempty" yaml:"useManagedIdentityExtension,omitempty"`
+	// UserAssignedIdentityID contains the Client ID of the user assigned MSI which is assigned to the underlying VMs. If empty the user assigned identity is not used.
+	// More details of the user assigned identity can be found at: https://docs.microsoft.com/en-us/azure/active-directory/managed-service-identity/overview
+	// For the user assigned identity specified here to be used, the UseManagedIdentityExtension has to be set to true.
+	UserAssignedIdentityID string `json:"userAssignedIdentityID,omitempty" yaml:"userAssignedIdentityID,omitempty"`
+	// The AAD Tenant ID for the Subscription that the network resources are deployed in
+	NetworkResourceTenantID string `json:"networkResourceTenantID,omitempty" yaml:"networkResourceTenantID,omitempty"`
+	// The AAD federated token file
+	AADFederatedTokenFile string `json:"aadFederatedTokenFile,omitempty" yaml:"aadFederatedTokenFile,omitempty"`
+	// Use workload identity federation for the virtual machine to access Azure ARM APIs
+	UseFederatedWorkloadIdentityExtension bool `json:"useFederatedWorkloadIdentityExtension,omitempty" yaml:"useFederatedWorkloadIdentityExtension,omitempty"`
+}
+
+var (
+	// ErrorNoAuth indicates that no credentials are provided.
+	ErrorNoAuth = fmt.Errorf("no credentials provided for Azure cloud provider")
+)
+
+type AuthProvider struct {
+	AzureAuthConfig
+	ARMClientConfig
+}
+
+const (
+	azureClientID           = "AZURE_CLIENT_ID"
+	azureFederatedTokenFile = "AZURE_FEDERATED_TOKEN_FILE"
+	azureTenantID           = "AZURE_TENANT_ID"
+)
+
+func NewAuthProvider(config AzureAuthConfig, armConfig ARMClientConfig) (*AuthProvider, error) {
+	// these environment variables are injected by workload identity webhook
+	if tenantID := os.Getenv(azureTenantID); tenantID != "" {
+		config.TenantID = tenantID
+	}
+	if clientID := os.Getenv(azureClientID); clientID != "" {
+		config.AADClientID = clientID
+	}
+	if federatedTokenFile := os.Getenv(azureFederatedTokenFile); federatedTokenFile != "" {
+		config.AADFederatedTokenFile = federatedTokenFile
+		config.UseFederatedWorkloadIdentityExtension = true
+	}
+	return &AuthProvider{
+		AzureAuthConfig: config,
+		ARMClientConfig: armConfig,
+	}, nil
+}
+
+func (factory *AuthProvider) GetAzIdentity() (azcore.TokenCredential, error) {
+	if factory.UseFederatedWorkloadIdentityExtension {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		return azidentity.NewWorkloadIdentityCredential(&azidentity.WorkloadIdentityCredentialOptions{
+			ClientOptions: clientOptions.ClientOptions,
+			ClientID:      factory.AADClientID,
+			TenantID:      factory.TenantID,
+			TokenFilePath: factory.AADFederatedTokenFile,
+		})
+	}
+
+	if factory.UseManagedIdentityExtension {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		credOptions := &azidentity.ManagedIdentityCredentialOptions{
+			ClientOptions: clientOptions.ClientOptions,
+		}
+		if len(factory.UserAssignedIdentityID) > 0 {
+			if strings.Contains(strings.ToUpper(factory.UserAssignedIdentityID), "/SUBSCRIPTIONS/") {
+				credOptions.ID = azidentity.ResourceID(factory.UserAssignedIdentityID)
+			} else {
+				credOptions.ID = azidentity.ClientID(factory.UserAssignedIdentityID)
+			}
+		}
+		return azidentity.NewManagedIdentityCredential(credOptions)
+	}
+
+	if len(factory.AADClientSecret) > 0 {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		credOptions := &azidentity.ClientSecretCredentialOptions{
+			ClientOptions: clientOptions.ClientOptions,
+		}
+		return azidentity.NewClientSecretCredential(factory.TenantID, factory.AADClientID, factory.AADClientSecret, credOptions)
+	}
+
+	if len(factory.AADClientCertPath) > 0 && len(factory.AADClientCertPassword) > 0 {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		credOptions := &azidentity.ClientCertificateCredentialOptions{
+			ClientOptions: clientOptions.ClientOptions,
+		}
+		certData, err := os.ReadFile(factory.AADClientCertPath)
+		if err != nil {
+			return nil, fmt.Errorf("reading the client certificate from file %s: %w", factory.AADClientCertPath, err)
+		}
+		certificate, privateKey, err := decodePkcs12(certData, factory.AADClientCertPassword)
+		if err != nil {
+			return nil, fmt.Errorf("decoding the client certificate: %w", err)
+		}
+		return azidentity.NewClientCertificateCredential(factory.TenantID, factory.AADClientID, []*x509.Certificate{certificate}, privateKey, credOptions)
+	}
+	return nil, ErrorNoAuth
+
+}
+
+// decodePkcs12 decodes a PKCS#12 client certificate by extracting the public certificate and
+// the private RSA key
+func decodePkcs12(pkcs []byte, password string) (*x509.Certificate, *rsa.PrivateKey, error) {
+	privateKey, certificate, err := pkcs12.Decode(pkcs, password)
+	if err != nil {
+		return nil, nil, fmt.Errorf("decoding the PKCS#12 client certificate: %w", err)
+	}
+	rsaPrivateKey, isRsaKey := privateKey.(*rsa.PrivateKey)
+	if !isRsaKey {
+		return nil, nil, fmt.Errorf("PKCS#12 certificate must contain a RSA private key")
+	}
+
+	return certificate, rsaPrivateKey, nil
+}
+
+func (factory *AuthProvider) GetNetworkAzIdentity() (azcore.TokenCredential, error) {
+	err := factory.checkConfigWhenNetworkResourceInDifferentTenant()
+	if err != nil {
+		return nil, fmt.Errorf("got error(%w) in getting network resources service principal token", err)
+	}
+	if len(factory.AADClientSecret) > 0 {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		credOptions := &azidentity.ClientSecretCredentialOptions{
+			ClientOptions: clientOptions.ClientOptions,
+		}
+		return azidentity.NewClientSecretCredential(factory.NetworkResourceTenantID, factory.AADClientID, factory.AADClientSecret, credOptions)
+	}
+	if len(factory.AADClientCertPath) > 0 && len(factory.AADClientCertPassword) > 0 {
+		return nil, fmt.Errorf("AAD Application client certificate authentication is not supported in getting network resources service principal token")
+	}
+	return nil, ErrorNoAuth
+}
+
+// UsesNetworkResourceInDifferentTenant determines whether the AzureAuthConfig indicates to use network resources in
+// different AAD Tenant than those for the cluster. Return true when NetworkResourceTenantID is specified  and not equal
+// to one defined in global configs
+func (factory *AuthProvider) UsesNetworkResourceInDifferentTenant() bool {
+	return len(factory.NetworkResourceTenantID) > 0 && !strings.EqualFold(factory.NetworkResourceTenantID, factory.TenantID)
+}
+
+// checkConfigWhenNetworkResourceInDifferentTenant checks configuration for the scenario of using network resource in different tenant
+func (factory *AuthProvider) checkConfigWhenNetworkResourceInDifferentTenant() error {
+	if !factory.UsesNetworkResourceInDifferentTenant() {
+		return fmt.Errorf("NetworkResourceTenantID must be configured")
+	}
+
+	if factory.UseManagedIdentityExtension {
+		return fmt.Errorf("managed identity is not supported")
+	}
+
+	return nil
+}
+
+func (factory *AuthProvider) GetMultiTenantIdentity() (azcore.TokenCredential, error) {
+	err := factory.checkConfigWhenNetworkResourceInDifferentTenant()
+	if err != nil {
+		return nil, fmt.Errorf("got error(%w) in getting network resources service principal token", err)
+	}
+
+	if len(factory.AADClientSecret) > 0 {
+		clientOptions, err := factory.GetDefaultClientOption()
+		if err != nil {
+			return nil, err
+		}
+		credOptions := &azidentity.ClientSecretCredentialOptions{
+			ClientOptions:              clientOptions.ClientOptions,
+			AdditionallyAllowedTenants: []string{factory.NetworkResourceTenantID},
+		}
+		return azidentity.NewClientSecretCredential(factory.TenantID, factory.AADClientID, factory.AADClientSecret, credOptions)
+	}
+	if len(factory.AADClientCertPath) > 0 && len(factory.AADClientCertPassword) > 0 {
+		return nil, fmt.Errorf("AAD Application client certificate authentication is not supported in getting multi-tenant service principal token")
+	}
+
+	return nil, ErrorNoAuth
+}
+
+func (factory *AuthProvider) GetDefaultClientOption() (*policy.ClientOptions, error) {
+
+	//Get default settings
+	options := utils.GetDefaultOption()
+
+	//update user agent header
+	options.ClientOptions.Telemetry.ApplicationID = factory.UserAgent
+	//todo: add backoff retry policy
+
+	//set cloud
+	var err error
+	options.ClientOptions.Cloud, err = AzureCloudConfigFromName(factory.Cloud, factory.ARMClientConfig.ResourceManagerEndpoint)
+	return options, err
+
+}
diff --git a/pkg/azclient/client-gen/generator/codegenerator.go b/pkg/azclient/client-gen/generator/codegenerator.go
index 763ec0200..1a121e025 100644
--- a/pkg/azclient/client-gen/generator/codegenerator.go
+++ b/pkg/azclient/client-gen/generator/codegenerator.go
@@ -18,6 +18,7 @@ limitations under the License.
 package generator
 
 import (
+	"fmt"
 	"go/ast"
 	"os/exec"
 
@@ -52,7 +53,8 @@ func (Generator) RegisterMarkers(into *markers.Registry) error {
 }
 
 func (g Generator) Generate(ctx *genall.GenerationContext) error {
-	if err := exec.Command("go", "get", "github.com/golang/mock/mockgen/model").Run(); err != nil {
+	if output, err := exec.Command("go", "get", "github.com/golang/mock/mockgen/model").CombinedOutput(); err != nil {
+		fmt.Println(string(output))
 		return err
 	}
 	var headerText string
@@ -95,11 +97,16 @@ func (g Generator) Generate(ctx *genall.GenerationContext) error {
 		}
 	}
 	//nolint:gosec // G204 ignore this!
-	if err := exec.Command("goimports", "-local", "sigs.k8s.io/cloud-provider-azure/pkg/azclient", "-w", ".").Run(); err != nil {
+	if output, err := exec.Command("goimports", "-local", "sigs.k8s.io/cloud-provider-azure/pkg/azclient", "-w", ".").CombinedOutput(); err != nil {
+		fmt.Println(string(output))
 		return err
 	}
 	//nolint:gosec // G204 ignore this!
-	return exec.Command("go", "test", "./...").Run()
+	if output, err := exec.Command("go", "test", "./...").CombinedOutput(); err != nil {
+		fmt.Println(string(output))
+		return err
+	}
+	return nil
 }
 
 func (Generator) CheckFilter() loader.NodeFilter {
diff --git a/pkg/azclient/config.go b/pkg/azclient/config.go
new file mode 100644
index 000000000..d6500cb01
--- /dev/null
+++ b/pkg/azclient/config.go
@@ -0,0 +1,72 @@
+/*
+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 azclient
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/cloud"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/policy/ratelimit"
+)
+
+type ClientFactoryConfig struct {
+	ratelimit.CloudProviderRateLimitConfig
+	ARMClientConfig
+
+	// Enable exponential backoff to manage resource request retries
+	CloudProviderBackoff bool `json:"cloudProviderBackoff,omitempty" yaml:"cloudProviderBackoff,omitempty"`
+}
+
+type ARMClientConfig struct {
+	// The cloud environment identifier. Takes values from https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/azcore@v1.6.0/cloud
+	Cloud string `json:"cloud,omitempty" yaml:"cloud,omitempty"`
+	// The user agent for Azure customer usage attribution
+	UserAgent string `json:"userAgent,omitempty" yaml:"userAgent,omitempty"`
+	// ResourceManagerEndpoint is the cloud's resource manager endpoint. If set, cloud provider queries this endpoint
+	// in order to generate an autorest.Environment instance instead of using one of the pre-defined Environments.
+	ResourceManagerEndpoint string `json:"resourceManagerEndpoint,omitempty" yaml:"resourceManagerEndpoint,omitempty"`
+}
+
+var EnvironmentMapping = map[string]cloud.Configuration{
+	"AZURECHINACLOUD":        cloud.AzureChina,
+	"AZURECLOUD":             cloud.AzurePublic,
+	"AZUREPUBLICCLOUD":       cloud.AzurePublic,
+	"AZUREUSGOVERNMENT":      cloud.AzureGovernment,
+	"AZUREUSGOVERNMENTCLOUD": cloud.AzureGovernment, //TODO: deprecate
+}
+
+var ConfigNotFoundError = errors.New("resource manager service config is not found")
+
+func AzureCloudConfigFromName(cloudName string, endpoint string) (cloud.Configuration, error) {
+	cloudName = strings.ToUpper(cloudName)
+	var cloudConfig cloud.Configuration
+	var ok bool
+	if cloudConfig, ok = EnvironmentMapping[cloudName]; !ok {
+		return cloudConfig, ConfigNotFoundError
+	}
+	if endpoint != "" {
+		var serviceConfig cloud.ServiceConfiguration
+		var ok bool
+		if serviceConfig, ok = cloudConfig.Services[cloud.ResourceManager]; !ok {
+			return cloudConfig, ConfigNotFoundError
+		}
+		serviceConfig.Endpoint = endpoint
+		cloudConfig.Services[cloud.ResourceManager] = serviceConfig
+	}
+	return cloudConfig, nil
+}
diff --git a/pkg/azclient/default.go b/pkg/azclient/default.go
new file mode 100644
index 000000000..f592cbcb5
--- /dev/null
+++ b/pkg/azclient/default.go
@@ -0,0 +1,42 @@
+/*
+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 azclient
+
+import (
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/arm/policy"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/utils"
+)
+
+func (factory *ClientFactory) GetDefaultResourceClientOption() (*policy.ClientOptions, error) {
+
+	//Get default settings
+	options := utils.GetDefaultOption()
+
+	//update user agent header
+	options.ClientOptions.Telemetry.ApplicationID = factory.ClientFactoryConfig.UserAgent
+
+	// add retry policy
+	if !factory.ClientFactoryConfig.CloudProviderBackoff {
+		options.ClientOptions.Retry.MaxRetries = 1
+	}
+
+	//set cloud
+	var err error
+	options.ClientOptions.Cloud, err = AzureCloudConfigFromName(factory.ClientFactoryConfig.Cloud, factory.ClientFactoryConfig.ResourceManagerEndpoint)
+	return options, err
+}
diff --git a/pkg/azclient/factory.go b/pkg/azclient/factory.go
new file mode 100644
index 000000000..a06277079
--- /dev/null
+++ b/pkg/azclient/factory.go
@@ -0,0 +1,63 @@
+/*
+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 azclient
+
+import (
+	"strings"
+	"sync"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/diskclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/policy/ratelimit"
+)
+
+type ClientFactory struct {
+	*ClientFactoryConfig
+	diskClientRegistry sync.Map
+	*AuthProvider
+}
+
+func NewClientFactory(config *ClientFactoryConfig, authProvider *AuthProvider) *ClientFactory {
+	return &ClientFactory{
+		ClientFactoryConfig: config,
+		AuthProvider:        authProvider,
+	}
+}
+
+func (factory *ClientFactory) GetDiskClient(subscription string) (diskclient.Interface, error) {
+	subID := strings.ToLower(subscription)
+
+	options, err := factory.GetDefaultResourceClientOption()
+	if err != nil {
+		return nil, err
+	}
+	//add ratelimit policy
+	ratelimitOption := factory.ClientFactoryConfig.GetRateLimitConfig("diskRateLimit")
+	rateLimitPolicy := ratelimit.NewRateLimitPolicy(ratelimitOption)
+	options.ClientOptions.PerCallPolicies = append(options.ClientOptions.PerCallPolicies, rateLimitPolicy)
+
+	cred, err := factory.AuthProvider.GetAzIdentity()
+	if err != nil {
+		return nil, err
+	}
+	defaultClient, err := diskclient.New(subscription, cred, options)
+	if err != nil {
+		return nil, err
+	}
+	client, _ := factory.diskClientRegistry.LoadOrStore(subID, &defaultClient)
+
+	return *client.(*diskclient.Interface), nil
+}
diff --git a/pkg/azclient/go.mod b/pkg/azclient/go.mod
index 82a076a18..8f8792a9d 100644
--- a/pkg/azclient/go.mod
+++ b/pkg/azclient/go.mod
@@ -18,9 +18,10 @@ require (
 	github.com/onsi/gomega v1.27.7
 	go.opentelemetry.io/otel v1.16.0
 	go.opentelemetry.io/otel/trace v1.16.0
+	golang.org/x/crypto v0.9.0
 	golang.org/x/time v0.3.0
 	gopkg.in/dnaeon/go-vcr.v3 v3.1.2
-	k8s.io/utils v0.0.0-20230209194617-a36077c30491
+	k8s.io/utils v0.0.0-20230505201702-9f6742963106
 )
 
 require (
@@ -30,18 +31,17 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
-	github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 // indirect
-	github.com/kr/pretty v0.3.0 // indirect
+	github.com/google/pprof v0.0.0-20230602010524-ada837c32108 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 // indirect
 	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	go.opentelemetry.io/otel/metric v1.16.0 // indirect
-	golang.org/x/crypto v0.8.0 // indirect
 	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/sys v0.8.0 // indirect
 	golang.org/x/text v0.9.0 // indirect
-	golang.org/x/tools v0.9.1 // indirect
-	google.golang.org/protobuf v1.28.1 // indirect
+	golang.org/x/tools v0.9.3 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/pkg/azclient/go.sum b/pkg/azclient/go.sum
index 247b168e8..f3c6f16cb 100644
--- a/pkg/azclient/go.sum
+++ b/pkg/azclient/go.sum
@@ -21,9 +21,6 @@ github.com/Azure/go-armbalancer v0.0.2 h1:NVnxsTWHI5/fEzL6k6TjxPUfcB/3Si3+HFOZXO
 github.com/Azure/go-armbalancer v0.0.2/go.mod h1:yTg7MA/8YnfKQc9o97tzAJ7fbdVkod1xGsIvKmhYPRE=
 github.com/AzureAD/microsoft-authentication-library-for-go v1.0.0 h1:OBhqkivkhkMqLPymWEppkm7vgPQY2XsHoEkaMQ0AdZY=
 github.com/AzureAD/microsoft-authentication-library-for-go v1.0.0/go.mod h1:kgDmCTgBzIEPFElEF+FK0SdjAor06dRq2Go927dnQ6o=
-github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
-github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
-github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -45,15 +42,13 @@ github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec=
-github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
+github.com/google/pprof v0.0.0-20230602010524-ada837c32108 h1:y+JfwMOPwQwIrnh3TUPwwtOAhONoppkHiSa4sQBoK2k=
+github.com/google/pprof v0.0.0-20230602010524-ada837c32108/go.mod h1:Jh3hGz2jkYak8qXPD19ryItVnUgpgeqzdkY/D0EaeuA=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
-github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
-github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -66,9 +61,10 @@ github.com/onsi/gomega v1.27.7 h1:fVih9JD6ogIiHUN6ePK7HJidyEDpWGVB5mzM7cWNXoU=
 github.com/onsi/gomega v1.27.7/go.mod h1:1p8OOlwo2iUUDsHnOrjE5UKYJ+e3W8eQ3qSlRahPmr4=
 github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 h1:KoWmjvw+nsYOo29YJK9vDA65RGE3NrOnUtO7a+RF9HU=
 github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8/go.mod h1:HKlIX3XHQyzLZPlr7++PzdhaXEj94dEiJgZDTsxEqUI=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -83,8 +79,8 @@ go.opentelemetry.io/otel/trace v1.16.0 h1:8JRpaObFoW0pxuVPapkgH8UhHQj+bJW8jJsCZE
 go.opentelemetry.io/otel/trace v1.16.0/go.mod h1:Yt9vYq1SdNz3xdjZZK7wcXv1qv2pwLkqr2QVwea0ef0=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.8.0 h1:pd9TJtTueMTVQXzk8E2XESSMQDj/U7OUu0PqJqPXQjQ=
-golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE=
+golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
+golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.10.0 h1:lFO9qtOdlre5W1jxS3r/4szv2/6iXxScdzjoBMXNhYk=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -96,7 +92,6 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -113,25 +108,23 @@ golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.9.1 h1:8WMNJAz3zrtPmnYC7ISf5dEn3MT0gY7jBJfw27yrrLo=
-golang.org/x/tools v0.9.1/go.mod h1:owI94Op576fPu3cIGQeHs3joujW/2Oc6MtlxbF5dfNc=
+golang.org/x/tools v0.9.3 h1:Gn1I8+64MsuTb/HpH+LmQtNas23LhUVr3rYZ0eKuaMM=
+golang.org/x/tools v0.9.3/go.mod h1:owI94Op576fPu3cIGQeHs3joujW/2Oc6MtlxbF5dfNc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
-google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/dnaeon/go-vcr.v3 v3.1.2 h1:F1smfXBqQqwpVifDfUBQG6zzaGjzT+EnVZakrOdr5wA=
 gopkg.in/dnaeon/go-vcr.v3 v3.1.2/go.mod h1:2IMOnnlx9I6u9x+YBsM3tAMx6AlOxnJ0pWxQAzZ79Ag=
-gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/utils v0.0.0-20230209194617-a36077c30491 h1:r0BAOLElQnnFhE/ApUsg3iHdVYYPBjNSSOMowRZxxsY=
-k8s.io/utils v0.0.0-20230209194617-a36077c30491/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
+k8s.io/utils v0.0.0-20230505201702-9f6742963106 h1:EObNQ3TW2D+WptiYXlApGNLVy0zm/JIBVY9i+M4wpAU=
+k8s.io/utils v0.0.0-20230505201702-9f6742963106/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
diff --git a/pkg/azclient/utils/flowcontrol/docs.go b/pkg/azclient/policy/ratelimit/flowcontrol/docs.go
similarity index 100%
rename from pkg/azclient/utils/flowcontrol/docs.go
rename to pkg/azclient/policy/ratelimit/flowcontrol/docs.go
diff --git a/pkg/azclient/utils/flowcontrol/throttle.go b/pkg/azclient/policy/ratelimit/flowcontrol/throttle.go
similarity index 100%
rename from pkg/azclient/utils/flowcontrol/throttle.go
rename to pkg/azclient/policy/ratelimit/flowcontrol/throttle.go
diff --git a/pkg/azclient/utils/flowcontrol/throttle_test.go b/pkg/azclient/policy/ratelimit/flowcontrol/throttle_test.go
similarity index 100%
rename from pkg/azclient/utils/flowcontrol/throttle_test.go
rename to pkg/azclient/policy/ratelimit/flowcontrol/throttle_test.go
diff --git a/pkg/azclient/policy/ratelimit.go b/pkg/azclient/policy/ratelimit/ratelimit.go
similarity index 65%
rename from pkg/azclient/policy/ratelimit.go
rename to pkg/azclient/policy/ratelimit/ratelimit.go
index 540b3d7bf..0ed37f258 100644
--- a/pkg/azclient/policy/ratelimit.go
+++ b/pkg/azclient/policy/ratelimit/ratelimit.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package policy
+package ratelimit
 
 import (
 	"errors"
@@ -22,11 +22,11 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 
-	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/utils/flowcontrol"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/policy/ratelimit/flowcontrol"
 )
 
-// RateLimitConfig indicates the rate limit config options.
-type RateLimitConfig struct {
+// Config indicates the rate limit config options.
+type Config struct {
 	// Enable rate limiting
 	CloudProviderRateLimit bool `json:"cloudProviderRateLimit,omitempty" yaml:"cloudProviderRateLimit,omitempty"`
 	// Rate limit QPS (Read)
@@ -39,7 +39,7 @@ type RateLimitConfig struct {
 	CloudProviderRateLimitBucketWrite int `json:"cloudProviderRateLimitBucketWrite,omitempty" yaml:"cloudProviderRateLimitBucketWrite,omitempty"`
 }
 
-func NewRateLimitPolicy(config *RateLimitConfig) policy.Policy {
+func NewRateLimitPolicy(config *Config) policy.Policy {
 	if config != nil && config.CloudProviderRateLimit {
 		readLimiter := flowcontrol.NewTokenBucketRateLimiter(
 			config.CloudProviderRateLimitQPS,
@@ -48,7 +48,7 @@ func NewRateLimitPolicy(config *RateLimitConfig) policy.Policy {
 		writeLimiter := flowcontrol.NewTokenBucketRateLimiter(
 			config.CloudProviderRateLimitQPSWrite,
 			config.CloudProviderRateLimitBucketWrite)
-		return &RateLimitPolicy{
+		return &Policy{
 			rateLimiterReader: readLimiter,
 			rateLimiterWriter: writeLimiter,
 		}
@@ -56,12 +56,12 @@ func NewRateLimitPolicy(config *RateLimitConfig) policy.Policy {
 	return nil
 }
 
-type RateLimitPolicy struct {
+type Policy struct {
 	rateLimiterWriter flowcontrol.RateLimiter
 	rateLimiterReader flowcontrol.RateLimiter
 }
 
-func (f RateLimitPolicy) Do(req *policy.Request) (*http.Response, error) {
+func (f Policy) Do(req *policy.Request) (*http.Response, error) {
 	if req.Raw().Method == http.MethodGet || req.Raw().Method == http.MethodHead {
 		if !f.rateLimiterReader.TryAccept() {
 			return nil, errors.New("rate limit reached")
@@ -73,3 +73,29 @@ func (f RateLimitPolicy) Do(req *policy.Request) (*http.Response, error) {
 	}
 	return req.Next()
 }
+
+// CloudProviderRateLimitConfig indicates the rate limit config for each clients.
+type CloudProviderRateLimitConfig struct {
+	// The default rate limit config options.
+	Config
+
+	// Rate limit config for each clients. Values would override default settings above.
+	Entries map[string]*Config `json:",inline" yaml:",inline"`
+}
+
+func NewCloudProviderRateLimitConfig() *CloudProviderRateLimitConfig {
+	return &CloudProviderRateLimitConfig{
+		Config: Config{
+			CloudProviderRateLimit: false,
+		},
+		Entries: make(map[string]*Config),
+	}
+}
+
+// GetRateLimitConfig returns the rate limit config for the given client. if the client is not found, the default is returned.
+func (config *CloudProviderRateLimitConfig) GetRateLimitConfig(clientName string) *Config {
+	if entry, ok := config.Entries[clientName]; ok {
+		return entry
+	}
+	return &config.Config
+}
diff --git a/pkg/azclient/policy/retry.go b/pkg/azclient/policy/retry.go
deleted file mode 100644
index efbcc71f4..000000000
--- a/pkg/azclient/policy/retry.go
+++ /dev/null
@@ -1,170 +0,0 @@
-/*
-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 policy
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io"
-	"math"
-	"net/http"
-	"strings"
-	"time"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
-)
-
-func GetRetriableStatusCode() []int {
-	return []int{
-		http.StatusRequestTimeout,      // 408
-		http.StatusInternalServerError, // 500
-		http.StatusBadGateway,          // 502
-		http.StatusServiceUnavailable,  // 503
-		http.StatusGatewayTimeout,      // 504
-	}
-}
-
-const (
-	DefaultMaxRetries    = 3
-	DefaultMaxRetryDelay = math.MaxInt64
-	DefaultRetryDelay    = 5 * time.Second
-	DefaultTryTimeout    = 1 * time.Second
-)
-
-type RetryConfig struct {
-	PollingDelay  *time.Duration
-	RetryAttempts *int
-
-	RetryDuration *time.Duration
-	// Backoff holds parameters applied to a Backoff function.
-	// The initial duration.
-	Duration time.Duration
-	// Duration is multiplied by factor each iteration, if factor is not zero
-	// and the limits imposed by Steps and Cap have not been reached.
-	// Should not be negative.
-	// The jitter does not contribute to the updates to the duration parameter.
-	Factor float64
-	// The sleep at each iteration is the duration plus an additional
-	// amount chosen uniformly at random from the interval between
-	// zero and `jitter*duration`.
-	Jitter float64
-	// The remaining number of iterations in which the duration
-	// parameter may change (but progress can be stopped earlier by
-	// hitting the cap). If not positive, the duration is not
-	// changed. Used for exponential backoff in combination with
-	// Factor and Cap.
-	Steps int
-	// A limit on revised values of the duration parameter. If a
-	// multiplication by the factor parameter would make the duration
-	// exceed the cap then the duration is set to the cap and the
-	// steps parameter is set to zero.
-	Cap time.Duration
-	// The errors indicate that the request shouldn't do more retrying.
-	NonRetriableErrors []string
-	// The RetriableHTTPStatusCodes indicates that the HTTPStatusCode should do more retrying.
-	RetriableHTTPStatusCodes []int
-}
-
-func NewGetRequestRegionalEndpointRetryPolicy(regionalEndpoint string) policy.Policy {
-	if regionalEndpoint == "" {
-		return nil
-	}
-	return &GetRequestRegionalEndpointRetryPolicy{
-		regionalEndpoint: regionalEndpoint,
-	}
-}
-
-// GetRequestRegionalEndpointRetryPolicy is a policy that retries requests on ResourceGroupNotFound error.
-type GetRequestRegionalEndpointRetryPolicy struct {
-	regionalEndpoint string
-}
-
-func (p *GetRequestRegionalEndpointRetryPolicy) Do(req *policy.Request) (*http.Response, error) {
-	response, err := req.Next()
-	if response == nil {
-		return response, err
-	}
-	//retry only when request is read operation and response is 404
-	if req.Raw().Method != http.MethodGet || req.Raw().Method != http.MethodHead {
-		return response, err
-	}
-	bodyBytes, _ := io.ReadAll(response.Body)
-	defer func() {
-		response.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
-	}()
-
-	bodyString := string(bodyBytes)
-	trimmed := strings.TrimSpace(bodyString)
-	// Hack: retry the regional ARM endpoint in case of ARM traffic split and arm resource group replication is too slow
-	// Empty content and 2xx http status code are returned in this case.
-	// Issue: https://github.com/kubernetes-sigs/cloud-provider-azure/issues/1296
-	// Such situation also needs retrying that ContentLength is -1, StatusCode is 200 and an empty body is returned.
-	emptyResp := (response.ContentLength == 0 || trimmed == "" || trimmed == "{}") && response.StatusCode >= 200 && response.StatusCode < 300
-	if !emptyResp {
-		if err == nil || response.StatusCode == http.StatusNotFound || p.regionalEndpoint == "" {
-			return response, err
-		}
-
-		var body map[string]interface{}
-		if e := json.Unmarshal(bodyBytes, &body); e != nil {
-			return response, err
-		}
-
-		errBody, ok := body["error"].(map[string]interface{})
-		if !ok || errBody["code"] == nil || !strings.EqualFold(errBody["code"].(string), "ResourceGroupNotFound") {
-			return response, err
-		}
-	}
-
-	// Do regional request
-	currentHost := req.Raw().URL.Host
-	if req.Raw().Host != "" {
-		currentHost = req.Raw().Host
-	}
-
-	if strings.HasPrefix(strings.ToLower(currentHost), strings.ToLower(p.regionalEndpoint)) {
-		return response, err
-	}
-
-	req.Raw().Host = p.regionalEndpoint
-	req.Raw().URL.Host = p.regionalEndpoint
-
-	regionalResponse, regionalError := req.Next()
-
-	// only use the result if the regional request actually goes through and returns 2xx status code, for two reasons:
-	// 1. the retry on regional ARM host approach is a hack.
-	// 2. the concatenated regional uri could be wrong as the rule is not officially declared by ARM.
-	if regionalResponse == nil || regionalResponse.StatusCode > 299 {
-		return response, err
-	}
-
-	// Do the same check on regional response just like the global one
-	bodyBytes, _ = io.ReadAll(regionalResponse.Body)
-	defer func() {
-		regionalResponse.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
-	}()
-	bodyString = string(bodyBytes)
-	trimmed = strings.TrimSpace(bodyString)
-	emptyResp = (regionalResponse.ContentLength == 0 || trimmed == "" || trimmed == "{}") && regionalResponse.StatusCode >= 200 && regionalResponse.StatusCode < 300
-	if emptyResp {
-		contentLengthErrStr := fmt.Sprintf("empty response with trimmed body %q, ContentLength %d and StatusCode %d", trimmed, regionalResponse.ContentLength, regionalResponse.StatusCode)
-		return response, fmt.Errorf(contentLengthErrStr)
-	}
-
-	return regionalResponse, regionalError
-}
diff --git a/pkg/azclient/policy/throttle.go b/pkg/azclient/policy/retryrepectthrottled/throttle.go
similarity index 89%
rename from pkg/azclient/policy/throttle.go
rename to pkg/azclient/policy/retryrepectthrottled/throttle.go
index d3f3dc075..80c5e33f8 100644
--- a/pkg/azclient/policy/throttle.go
+++ b/pkg/azclient/policy/retryrepectthrottled/throttle.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package policy
+package retryrepectthrottled
 
 import (
 	"errors"
@@ -35,6 +35,16 @@ func NewThrottlingPolicy() policy.Policy {
 	}
 }
 
+func GetRetriableStatusCode() []int {
+	return []int{
+		http.StatusRequestTimeout,      // 408
+		http.StatusInternalServerError, // 500
+		http.StatusBadGateway,          // 502
+		http.StatusServiceUnavailable,  // 503
+		http.StatusGatewayTimeout,      // 504
+	}
+}
+
 // ThrottlingPolicy implements the Azure SDK for Go's Policy interface.
 // throttle counter is based on resources operation per subscription.
 type ThrottlingPolicy struct {
diff --git a/pkg/azclient/utils/options.go b/pkg/azclient/utils/options.go
index 0d2a80707..76cdfe6f0 100644
--- a/pkg/azclient/utils/options.go
+++ b/pkg/azclient/utils/options.go
@@ -17,25 +17,35 @@ limitations under the License.
 package utils
 
 import (
+	"math"
+	"time"
+
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/tracing"
 
-	custompolicy "sigs.k8s.io/cloud-provider-azure/pkg/azclient/policy"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/policy/retryrepectthrottled"
+)
+
+const (
+	DefaultMaxRetries    = 3
+	DefaultMaxRetryDelay = math.MaxInt64
+	DefaultRetryDelay    = 5 * time.Second
+	DefaultTryTimeout    = 1 * time.Second
 )
 
 func GetDefaultOption() *arm.ClientOptions {
 	return &arm.ClientOptions{
 		ClientOptions: policy.ClientOptions{
 			Retry: policy.RetryOptions{
-				RetryDelay:    custompolicy.DefaultRetryDelay,
-				MaxRetryDelay: custompolicy.DefaultMaxRetryDelay,
-				MaxRetries:    custompolicy.DefaultMaxRetries,
-				TryTimeout:    custompolicy.DefaultTryTimeout,
-				StatusCodes:   custompolicy.GetRetriableStatusCode(),
+				RetryDelay:    DefaultRetryDelay,
+				MaxRetryDelay: DefaultMaxRetryDelay,
+				MaxRetries:    DefaultMaxRetries,
+				TryTimeout:    DefaultTryTimeout,
+				StatusCodes:   retryrepectthrottled.GetRetriableStatusCode(),
 			},
 			PerRetryPolicies: []policy.Policy{
-				custompolicy.NewThrottlingPolicy(),
+				retryrepectthrottled.NewThrottlingPolicy(),
 			},
 			Transport: defaultHTTPClient,
 			TracingProvider: tracing.NewProvider(func(name, version string) tracing.Tracer {
diff --git a/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient.go b/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient.go
index f641d2b44..5c41f79b8 100644
--- a/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient.go
+++ b/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient.go
@@ -55,10 +55,6 @@ type Client struct {
 	RetryAfterWriter time.Time
 }
 
-type backendPoolsToBeMigrated struct {
-	BackendPoolNames []string `json:"pools"`
-}
-
 // New creates a new LoadBalancer client with ratelimiting.
 func New(config *azclients.ClientConfig) *Client {
 	baseURI := config.ResourceManagerEndpoint
@@ -425,69 +421,6 @@ func (page LoadBalancerListResultPage) Values() []network.LoadBalancer {
 	return *page.lblr.Value
 }
 
-// GetLBBackendPool gets a LoadBalancer backend pool.
-func (c *Client) GetLBBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolName string, expand string) (network.BackendAddressPool, *retry.Error) {
-	mc := metrics.NewMetricContext("load_balancers", "get_backend_pool", resourceGroupName, c.subscriptionID, "")
-
-	// Report errors if the client is rate limited.
-	if !c.rateLimiterReader.TryAccept() {
-		mc.RateLimitedCount()
-		return network.BackendAddressPool{}, retry.GetRateLimitError(false, "LBGet")
-	}
-
-	// Report errors if the client is throttled.
-	if c.RetryAfterReader.After(time.Now()) {
-		mc.ThrottledCount()
-		rerr := retry.GetThrottlingError("LBBackendPoolGet", "client throttled", c.RetryAfterReader)
-		return network.BackendAddressPool{}, rerr
-	}
-
-	result, rerr := c.getLBBackendPool(ctx, resourceGroupName, loadBalancerName, backendPoolName, 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
-}
-
-// getLBBackendPool gets a LoadBalancer backend pool.
-func (c *Client) getLBBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolName string, expand string) (network.BackendAddressPool, *retry.Error) {
-	resourceID := armclient.GetChildResourceID(
-		c.subscriptionID,
-		resourceGroupName,
-		lbResourceType,
-		loadBalancerName,
-		"backendAddressPools",
-		backendPoolName,
-	)
-	result := network.BackendAddressPool{}
-
-	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", "loadbalancer.backendpool.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", "loadbalancer.backendpool.get.respond", resourceID, err)
-		return result, retry.GetError(response, err)
-	}
-
-	result.Response = autorest.Response{Response: response}
-	return result, nil
-}
-
 // CreateOrUpdateBackendPools creates or updates a LoadBalancer backend pool.
 func (c *Client) CreateOrUpdateBackendPools(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolName string, parameters network.BackendAddressPool, etag string) *retry.Error {
 	mc := metrics.NewMetricContext("load_balancers", "create_or_update_backend_pools", resourceGroupName, c.subscriptionID, "")
@@ -537,14 +470,14 @@ func (c *Client) createOrUpdateLBBackendPool(ctx context.Context, resourceGroupN
 	response, rerr := c.armClient.PutResource(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", "loadbalancer.backendpool.put.request", resourceID, rerr.Error())
+		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "loadbalancerbackendpool.put.request", resourceID, rerr.Error())
 		return rerr
 	}
 
 	if response != nil && response.StatusCode != http.StatusNoContent {
 		_, rerr = c.createOrUpdateBackendPoolResponder(response)
 		if rerr != nil {
-			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "loadbalancer.backendpool.put.respond", resourceID, rerr.Error())
+			klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "loadbalancerbackendpool.put.respond", resourceID, rerr.Error())
 			return rerr
 		}
 	}
@@ -604,56 +537,3 @@ func (c *Client) createOrUpdateBackendPoolResponder(resp *http.Response) (*netwo
 	result.Response = autorest.Response{Response: resp}
 	return result, retry.GetError(resp, err)
 }
-
-// MigrateToIPBasedBackendPool migrates a NIC-based backend pool to IP-based.
-func (c *Client) MigrateToIPBasedBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolNames []string) *retry.Error {
-	mc := metrics.NewMetricContext("load_balancers", "migrate_to_ip_based_backend_pool", resourceGroupName, c.subscriptionID, "")
-
-	// Report errors if the client is rate limited.
-	if !c.rateLimiterWriter.TryAccept() {
-		mc.RateLimitedCount()
-		return retry.GetRateLimitError(true, "LBMigrateToIPBasedBackendPool")
-	}
-
-	// Report errors if the client is throttled.
-	if c.RetryAfterWriter.After(time.Now()) {
-		mc.ThrottledCount()
-		rerr := retry.GetThrottlingError("LBMigrateToIPBasedBackendPool", "client throttled", c.RetryAfterWriter)
-		return rerr
-	}
-
-	parameters := backendPoolsToBeMigrated{
-		BackendPoolNames: backendPoolNames,
-	}
-	rerr := c.migrateToIPBasedBackendPool(ctx, resourceGroupName, loadBalancerName, parameters)
-	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) migrateToIPBasedBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, parameters backendPoolsToBeMigrated) *retry.Error {
-	resourceID := armclient.GetResourceID(
-		c.subscriptionID,
-		resourceGroupName,
-		lbResourceType,
-		loadBalancerName,
-	)
-
-	response, rerr := c.armClient.PostResource(ctx, resourceID, "migrateToIpBased", parameters, map[string]interface{}{})
-	defer c.armClient.CloseResponse(ctx, response)
-	if rerr != nil {
-		klog.V(5).Infof("Received error in %s: resourceID: %s, error: %s", "loadbalancerbackendpool.migrate.request", resourceID, rerr.Error())
-		return rerr
-	}
-
-	klog.Infof("Response: %v", response)
-	return nil
-}
diff --git a/pkg/azureclients/loadbalancerclient/interface.go b/pkg/azureclients/loadbalancerclient/interface.go
index 6f0feb22f..7a1a88f7d 100644
--- a/pkg/azureclients/loadbalancerclient/interface.go
+++ b/pkg/azureclients/loadbalancerclient/interface.go
@@ -51,12 +51,6 @@ type Interface interface {
 	// Delete deletes a LoadBalancer by name.
 	Delete(ctx context.Context, resourceGroupName string, loadBalancerName string) *retry.Error
 
-	// GetLBBackendPool gets a LoadBalancer backend pool.
-	GetLBBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolName string, expand string) (network.BackendAddressPool, *retry.Error)
-
 	// DeleteLBBackendPool deletes a LoadBalancer backend pool by name.
 	DeleteLBBackendPool(ctx context.Context, resourceGroupName, loadBalancerName, backendPoolName string) *retry.Error
-
-	// MigrateToIPBasedBackendPool migrates a NIC-based backend pool to IP-based.
-	MigrateToIPBasedBackendPool(ctx context.Context, resourceGroupName string, loadBalancerName string, backendPoolNames []string) *retry.Error
 }
diff --git a/pkg/azureclients/loadbalancerclient/mockloadbalancerclient/interface.go b/pkg/azureclients/loadbalancerclient/mockloadbalancerclient/interface.go
index c32c8a230..9397a9782 100644
--- a/pkg/azureclients/loadbalancerclient/mockloadbalancerclient/interface.go
+++ b/pkg/azureclients/loadbalancerclient/mockloadbalancerclient/interface.go
@@ -124,21 +124,6 @@ func (mr *MockInterfaceMockRecorder) Get(ctx, resourceGroupName, loadBalancerNam
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockInterface)(nil).Get), ctx, resourceGroupName, loadBalancerName, expand)
 }
 
-// GetLBBackendPool mocks base method.
-func (m *MockInterface) GetLBBackendPool(ctx context.Context, resourceGroupName, loadBalancerName, backendPoolName, expand string) (network.BackendAddressPool, *retry.Error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetLBBackendPool", ctx, resourceGroupName, loadBalancerName, backendPoolName, expand)
-	ret0, _ := ret[0].(network.BackendAddressPool)
-	ret1, _ := ret[1].(*retry.Error)
-	return ret0, ret1
-}
-
-// GetLBBackendPool indicates an expected call of GetLBBackendPool.
-func (mr *MockInterfaceMockRecorder) GetLBBackendPool(ctx, resourceGroupName, loadBalancerName, backendPoolName, expand interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetLBBackendPool", reflect.TypeOf((*MockInterface)(nil).GetLBBackendPool), ctx, resourceGroupName, loadBalancerName, backendPoolName, expand)
-}
-
 // List mocks base method.
 func (m *MockInterface) List(ctx context.Context, resourceGroupName string) ([]network.LoadBalancer, *retry.Error) {
 	m.ctrl.T.Helper()
@@ -153,17 +138,3 @@ func (mr *MockInterfaceMockRecorder) List(ctx, resourceGroupName interface{}) *g
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockInterface)(nil).List), ctx, resourceGroupName)
 }
-
-// MigrateToIPBasedBackendPool mocks base method.
-func (m *MockInterface) MigrateToIPBasedBackendPool(ctx context.Context, resourceGroupName, loadBalancerName string, backendPoolNames []string) *retry.Error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "MigrateToIPBasedBackendPool", ctx, resourceGroupName, loadBalancerName, backendPoolNames)
-	ret0, _ := ret[0].(*retry.Error)
-	return ret0
-}
-
-// MigrateToIPBasedBackendPool indicates an expected call of MigrateToIPBasedBackendPool.
-func (mr *MockInterfaceMockRecorder) MigrateToIPBasedBackendPool(ctx, resourceGroupName, loadBalancerName, backendPoolNames interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MigrateToIPBasedBackendPool", reflect.TypeOf((*MockInterface)(nil).MigrateToIPBasedBackendPool), ctx, resourceGroupName, loadBalancerName, backendPoolNames)
-}
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index b49604465..50c15f683 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -185,6 +185,16 @@ type Config struct {
 	// Use instance metadata service where possible
 	UseInstanceMetadata bool `json:"useInstanceMetadata,omitempty" yaml:"useInstanceMetadata,omitempty"`
 
+	// EnableMultipleStandardLoadBalancers determines the behavior of the standard load balancer. If set to true
+	// there would be one standard load balancer per VMAS or VMSS, which is similar with the behavior of the basic
+	// load balancer. Users could select the specific standard load balancer for their service by the service
+	// annotation `service.beta.kubernetes.io/azure-load-balancer-mode`, If set to false, the same standard load balancer
+	// would be shared by all services in the cluster. In this case, the mode selection annotation would be ignored.
+	EnableMultipleStandardLoadBalancers bool `json:"enableMultipleStandardLoadBalancers,omitempty" yaml:"enableMultipleStandardLoadBalancers,omitempty"`
+	// NodePoolsWithoutDedicatedSLB stores the VMAS/VMSS names that share the primary standard load balancer instead
+	// of having a dedicated one. This is useful only when EnableMultipleStandardLoadBalancers is set to true.
+	NodePoolsWithoutDedicatedSLB string `json:"nodePoolsWithoutDedicatedSLB,omitempty" yaml:"nodePoolsWithoutDedicatedSLB,omitempty"`
+
 	// Backoff exponent
 	CloudProviderBackoffExponent float64 `json:"cloudProviderBackoffExponent,omitempty" yaml:"cloudProviderBackoffExponent,omitempty"`
 	// Backoff jitter
@@ -249,12 +259,6 @@ type Config struct {
 	PutVMSSVMBatchSize int `json:"putVMSSVMBatchSize" yaml:"putVMSSVMBatchSize"`
 	// PrivateLinkServiceResourceGroup determines the specific resource group of the private link services user want to use
 	PrivateLinkServiceResourceGroup string `json:"privateLinkServiceResourceGroup,omitempty" yaml:"privateLinkServiceResourceGroup,omitempty"`
-
-	// EnableMigrateToIPBasedBackendPoolAPI uses the migration API to migrate from NIC-based to IP-based backend pool.
-	// The migration API can provide a migration from NIC-based to IP-based backend pool without service downtime.
-	// If the API is not used, the migration will be done by decoupling all nodes on the backend pool and then re-attaching
-	// node IPs, which will introduce service downtime. The downtime increases with the number of nodes in the backend pool.
-	EnableMigrateToIPBasedBackendPoolAPI bool `json:"enableMigrateToIPBasedBackendPoolAPI" yaml:"enableMigrateToIPBasedBackendPoolAPI"`
 }
 
 type InitSecretConfig struct {
@@ -316,6 +320,8 @@ type Cloud struct {
 
 	// ipv6DualStack allows overriding for unit testing.  It's normally initialized from featuregates
 	ipv6DualStackEnabled bool
+	// isSHaredLoadBalancerSynced indicates if the reconcileSharedLoadBalancer has been run
+	isSharedLoadBalancerSynced bool
 	// Lock for access to node caches, includes nodeZones, nodeResourceGroups, and unmanagedNodes.
 	nodeCachesLock sync.RWMutex
 	// nodeNames holds current nodes for tracking added nodes in VM caches.
diff --git a/pkg/provider/azure_backoff.go b/pkg/provider/azure_backoff.go
index c351c1054..279a88501 100644
--- a/pkg/provider/azure_backoff.go
+++ b/pkg/provider/azure_backoff.go
@@ -17,14 +17,12 @@ limitations under the License.
 package provider
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
 	"regexp"
 	"strings"
-	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-08-01/compute"
 	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2022-07-01/network"
@@ -654,54 +652,3 @@ func (az *Cloud) CreateOrUpdateSubnet(service *v1.Service, subnet network.Subnet
 
 	return nil
 }
-
-// MigrateToIPBasedBackendPoolAndWaitForCompletion use the migration API to migrate from
-// NIC-based to IP-based LB backend pools. It also makes sure the number of IP addresses
-// in the backend pools is expected.
-func (az *Cloud) MigrateToIPBasedBackendPoolAndWaitForCompletion(
-	lbName string, backendPoolNames []string, nicsCountMap map[string]int,
-) error {
-	if rerr := az.LoadBalancerClient.MigrateToIPBasedBackendPool(context.Background(), az.ResourceGroup, lbName, backendPoolNames); rerr != nil {
-		backendPoolNamesStr := strings.Join(backendPoolNames, ",")
-		klog.Errorf("MigrateToIPBasedBackendPoolAndWaitForCompletion: Failed to migrate to IP based backend pool for lb %s, backend pool %s: %s", lbName, backendPoolNamesStr, rerr.Error().Error())
-		return rerr.Error()
-	}
-
-	succeeded := make(map[string]bool)
-	for bpName := range nicsCountMap {
-		succeeded[bpName] = false
-	}
-
-	err := wait.PollImmediate(5*time.Second, 10*time.Minute, func() (done bool, err error) {
-		for bpName, nicsCount := range nicsCountMap {
-			if succeeded[bpName] {
-				continue
-			}
-
-			bp, rerr := az.LoadBalancerClient.GetLBBackendPool(context.Background(), az.ResourceGroup, lbName, bpName, "")
-			if rerr != nil {
-				klog.Errorf("MigrateToIPBasedBackendPoolAndWaitForCompletion: Failed to get backend pool %s for lb %s: %s", bpName, lbName, rerr.Error().Error())
-				return false, rerr.Error()
-			}
-
-			if countIPsOnBackendPool(bp) != nicsCount {
-				klog.V(4).Infof("MigrateToIPBasedBackendPoolAndWaitForCompletion: Expected IPs %s, current IPs %d, will retry in 5s", nicsCount, countIPsOnBackendPool(bp))
-				return false, nil
-			}
-			succeeded[bpName] = true
-		}
-		return true, nil
-	})
-
-	if err != nil {
-		if errors.Is(err, wait.ErrWaitTimeout) {
-			klog.Warningf("MigrateToIPBasedBackendPoolAndWaitForCompletion: Timeout waiting for migration to IP based backend pool for lb %s, backend pool %s", lbName, strings.Join(backendPoolNames, ","))
-			return nil
-		}
-
-		klog.Errorf("MigrateToIPBasedBackendPoolAndWaitForCompletion: Failed to wait for migration to IP based backend pool for lb %s, backend pool %s: %s", lbName, strings.Join(backendPoolNames, ","), err.Error())
-		return err
-	}
-
-	return nil
-}
diff --git a/pkg/provider/azure_loadbalancer.go b/pkg/provider/azure_loadbalancer.go
index efb79f318..05f19a623 100644
--- a/pkg/provider/azure_loadbalancer.go
+++ b/pkg/provider/azure_loadbalancer.go
@@ -336,7 +336,8 @@ func (az *Cloud) shouldChangeLoadBalancer(service *v1.Service, currLBName, clust
 	}
 
 	// if using the single standard load balancer, the current LB should be kept
-	if az.useStandardLoadBalancer() {
+	useSingleSLB := az.useStandardLoadBalancer() && !az.EnableMultipleStandardLoadBalancers
+	if useSingleSLB {
 		return false
 	}
 
@@ -354,6 +355,15 @@ func (az *Cloud) shouldChangeLoadBalancer(service *v1.Service, currLBName, clust
 		return false
 	}
 
+	// if the vmSet selected by the annotation is sharing the primary slb, and the service
+	// has been associated to the primary slb, keep it
+	useMultipleSLBs := az.useStandardLoadBalancer() && az.EnableMultipleStandardLoadBalancers
+	if useMultipleSLBs &&
+		az.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vmSetName)) &&
+		strings.EqualFold(lbName, clusterName) {
+		return false
+	}
+
 	// if the VMSS/VMAS of the current LB is different from the mode, change the LB
 	// to another one
 	klog.V(2).Infof("shouldChangeLoadBalancer(%s, %s, %s): change the LB to another one", service.Name, currLBName, clusterName)
@@ -513,6 +523,91 @@ func (az *Cloud) safeDeleteLoadBalancer(lb network.LoadBalancer, clusterName, vm
 	return nil
 }
 
+// reconcileSharedLoadBalancer deletes the dedicated SLBs of the non-primary vmSets. There are
+// two scenarios where this operation is needed:
+// 1. Using multiple slbs and the vmSet is supposed to share the primary slb.
+// 2. When migrating from multiple slbs to single slb mode.
+// It also ensures those vmSets are joint the backend pools of the primary SLBs.
+// It runs only once every time the cloud controller manager restarts.
+func (az *Cloud) reconcileSharedLoadBalancer(service *v1.Service, clusterName string, nodes []*v1.Node) ([]network.LoadBalancer, error) {
+	var (
+		existingLBs []network.LoadBalancer
+		err         error
+	)
+
+	existingLBs, err = az.ListManagedLBs(service, nodes, clusterName)
+	if err != nil {
+		return nil, fmt.Errorf("reconcileSharedLoadBalancer: failed to list managed LB: %w", err)
+	}
+
+	// only run once since the controller manager rebooted
+	if az.isSharedLoadBalancerSynced {
+		return existingLBs, nil
+	}
+	defer func() {
+		if err == nil {
+			az.isSharedLoadBalancerSynced = true
+		}
+	}()
+
+	// skip if the cluster is using basic LB
+	if !az.useStandardLoadBalancer() {
+		return existingLBs, nil
+	}
+
+	// Skip if nodes is nil, which means the service is being deleted.
+	// When nodes is nil, all LBs included unmanaged LBs will be returned,
+	// if we don't skip this function, the unmanaged ones may be deleted later.
+	if nodes == nil {
+		klog.V(4).Infof("reconcileSharedLoadBalancer: returning early because the service %s is being deleted", service.Name)
+		return existingLBs, nil
+	}
+
+	lbNamesToBeDeleted := sets.New[string]()
+	// delete unwanted LBs
+	for _, lb := range existingLBs {
+		klog.V(4).Infof("reconcileSharedLoadBalancer: checking LB %s", pointer.StringDeref(lb.Name, ""))
+		// skip the internal or external primary load balancer
+		lbNamePrefix := strings.TrimSuffix(pointer.StringDeref(lb.Name, ""), consts.InternalLoadBalancerNameSuffix)
+		if strings.EqualFold(lbNamePrefix, clusterName) {
+			continue
+		}
+
+		// skip if the multiple slbs mode is enabled and
+		// the vmSet is supposed to have dedicated SLBs
+		vmSetName := strings.ToLower(az.mapLoadBalancerNameToVMSet(pointer.StringDeref(lb.Name, ""), clusterName))
+		if az.EnableMultipleStandardLoadBalancers && !az.getVMSetNamesSharingPrimarySLB().Has(vmSetName) {
+			klog.V(4).Infof("reconcileSharedLoadBalancer: skip deleting the LB %s because the vmSet %s needs a dedicated SLB", pointer.StringDeref(lb.Name, ""), vmSetName)
+			continue
+		}
+
+		// For non-primary load balancer, the lb name is the name of the VMSet.
+		// If the VMSet name is in az.NodePoolsWithoutDedicatedSLB, we should
+		// decouple the VMSet from the lb and delete the lb. Then adding the VMSet
+		// to the backend pool of the primary slb.
+		klog.V(2).Infof("reconcileSharedLoadBalancer: deleting LB %s because the corresponding vmSet is supposed to be in the primary SLB", pointer.StringDeref(lb.Name, ""))
+		rerr := az.safeDeleteLoadBalancer(lb, clusterName, vmSetName, service)
+		if rerr != nil {
+			return nil, rerr.Error()
+		}
+
+		// remove the deleted lb from the list and construct a new primary
+		// lb, so that getServiceLoadBalancer doesn't have to call list api again
+		lbNamesToBeDeleted.Insert(strings.ToLower(pointer.StringDeref(lb.Name, "")))
+	}
+
+	for i := len(existingLBs) - 1; i >= 0; i-- {
+		lb := existingLBs[i]
+		if !lbNamesToBeDeleted.Has(strings.ToLower(pointer.StringDeref(lb.Name, ""))) {
+			continue
+		}
+		// remove the deleted LB from the list
+		existingLBs = append(existingLBs[:i], existingLBs[i+1:]...)
+	}
+
+	return existingLBs, nil
+}
+
 // getServiceLoadBalancer gets the loadbalancer for the service if it already exists.
 // If wantLb is TRUE then -it selects a new load balancer.
 // In case the selected load balancer does not exist it returns network.LoadBalancer struct
@@ -523,6 +618,7 @@ func (az *Cloud) getServiceLoadBalancer(service *v1.Service, clusterName string,
 	var defaultLB *network.LoadBalancer
 	primaryVMSetName := az.VMSet.GetPrimaryVMSetName()
 	defaultLBName := az.getAzureLoadBalancerName(clusterName, primaryVMSetName, isInternal)
+	useMultipleSLBs := az.useStandardLoadBalancer() && az.EnableMultipleStandardLoadBalancers
 
 	// reuse the lb list from reconcileSharedLoadBalancer to reduce the api call
 	if len(existingLBs) == 0 {
@@ -535,7 +631,32 @@ func (az *Cloud) getServiceLoadBalancer(service *v1.Service, clusterName string,
 	// check if the service already has a load balancer
 	for i := range existingLBs {
 		existingLB := existingLBs[i]
+		existingLBNamePrefix := strings.TrimSuffix(pointer.StringDeref(existingLB.Name, ""), consts.InternalLoadBalancerNameSuffix)
+
+		// for the primary standard load balancer (internal or external), when enabled multiple slbs
+		if strings.EqualFold(existingLBNamePrefix, clusterName) && useMultipleSLBs {
+			shouldRemoveVMSetFromSLB := func(vmSetName string) bool {
+				// not removing the vmSet from the primary SLB
+				// if it is supposed to share the primary SLB.
+				if az.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vmSetName)) {
+					return false
+				}
 
+				// removing the vmSet from the primary SLB if
+				// it is not the primary vmSet. There are two situations:
+				// 1. when migrating from single SLB to multiple SLBs, we
+				// need to remove all non-primary vmSets from the primary SLB;
+				// 2. when migrating from shared mode to dedicated SLB, we
+				// need to remove the specific vmSet from the primary SLB.
+				return !strings.EqualFold(vmSetName, primaryVMSetName) && vmSetName != ""
+			}
+			cleanedLB, err := az.LoadBalancerBackendPool.CleanupVMSetFromBackendPoolByCondition(&existingLB, service, nodes, clusterName, shouldRemoveVMSetFromSLB)
+			if err != nil {
+				return nil, nil, false, err
+			}
+			existingLB = *cleanedLB
+			existingLBs[i] = *cleanedLB
+		}
 		if strings.EqualFold(*existingLB.Name, defaultLBName) {
 			defaultLB = &existingLB
 		}
@@ -569,7 +690,8 @@ func (az *Cloud) getServiceLoadBalancer(service *v1.Service, clusterName string,
 	// Service does not have a load balancer, select one.
 	// Single standard load balancer doesn't need this because
 	// all backends nodes should be added to same LB.
-	if wantLb && !az.useStandardLoadBalancer() {
+	useSingleSLB := az.useStandardLoadBalancer() && !az.EnableMultipleStandardLoadBalancers
+	if wantLb && !useSingleSLB {
 		// select new load balancer for service
 		selectedLB, exists, err := az.selectLoadBalancer(clusterName, service, &existingLBs, nodes)
 		if err != nil {
@@ -1431,9 +1553,10 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 	serviceName := getServiceName(service)
 	klog.V(2).Infof("reconcileLoadBalancer for service(%s) - wantLb(%t): started", serviceName, wantLb)
 
-	existingLBs, err := az.ListManagedLBs(service, nodes, clusterName)
+	existingLBs, err := az.reconcileSharedLoadBalancer(service, clusterName, nodes)
 	if err != nil {
-		return nil, fmt.Errorf("reconcileLoadBalancer: failed to list managed LB: %w", err)
+		klog.Errorf("reconcileLoadBalancer: failed to reconcile shared load balancer: %v", err)
+		return nil, err
 	}
 
 	lb, lbStatus, _, err := az.getServiceLoadBalancer(service, clusterName, nodes, wantLb, existingLBs)
@@ -1456,7 +1579,7 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 
 	// reconcile the load balancer's backend pool configuration.
 	if wantLb {
-		preConfig, changed, shouldRefreshLB, err := az.LoadBalancerBackendPool.ReconcileBackendPools(clusterName, service, lb)
+		preConfig, changed, err := az.LoadBalancerBackendPool.ReconcileBackendPools(clusterName, service, lb)
 		if err != nil {
 			return lb, err
 		}
@@ -1464,16 +1587,6 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 			dirtyLb = true
 		}
 		isBackendPoolPreConfigured = preConfig
-
-		// If the LB is changed, refresh it to avoid etag mismatch error
-		// later when create or update the LB.
-		if shouldRefreshLB {
-			klog.V(4).Infof("reconcileLoadBalancer for service(%s): refreshing load balancer %s", serviceName, lbName)
-			lb, _, err = az.getAzureLoadBalancer(lbName, azcache.CacheReadTypeForceRefresh)
-			if err != nil {
-				return lb, fmt.Errorf("reconcileLoadBalancer for service (%s): failed to get load balancer %s: %w", serviceName, lbName, err)
-			}
-		}
 	}
 
 	// reconcile the load balancer's frontend IP configurations.
@@ -3498,7 +3611,8 @@ func ipInSubnet(ip string, subnet *network.Subnet) bool {
 // if anything else it returns the unique VM set names after trimming spaces.
 func (az *Cloud) getServiceLoadBalancerMode(service *v1.Service) (bool, bool, string) {
 	mode, hasMode := service.Annotations[consts.ServiceAnnotationLoadBalancerMode]
-	if az.useStandardLoadBalancer() && hasMode {
+	useSingleSLB := az.useStandardLoadBalancer() && !az.EnableMultipleStandardLoadBalancers
+	if useSingleSLB && hasMode {
 		klog.Warningf("single standard load balancer doesn't work with annotation %q, would ignore it", consts.ServiceAnnotationLoadBalancerMode)
 	}
 	mode = strings.TrimSpace(mode)
diff --git a/pkg/provider/azure_loadbalancer_backendpool.go b/pkg/provider/azure_loadbalancer_backendpool.go
index 650e102b5..993048b94 100644
--- a/pkg/provider/azure_loadbalancer_backendpool.go
+++ b/pkg/provider/azure_loadbalancer_backendpool.go
@@ -33,6 +33,7 @@ import (
 	"k8s.io/utils/pointer"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/cache"
+	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
 	"sigs.k8s.io/cloud-provider-azure/pkg/metrics"
 )
 
@@ -51,7 +52,7 @@ type BackendPool interface {
 
 	// ReconcileBackendPools creates the inbound backend pool if it is not existed, and removes nodes that are supposed to be
 	// excluded from the load balancers.
-	ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, bool, error)
+	ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, error)
 
 	// GetBackendPrivateIPs returns the private IPs of LoadBalancer's backend pool
 	GetBackendPrivateIPs(clusterName string, service *v1.Service, lb *network.LoadBalancer) ([]string, []string)
@@ -156,7 +157,7 @@ func (bc *backendPoolTypeNodeIPConfig) CleanupVMSetFromBackendPoolByCondition(sl
 	return slb, nil
 }
 
-func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, bool, error) {
+func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, error) {
 	var newBackendPools []network.BackendAddressPool
 	var err error
 	if lb.BackendAddressPools != nil {
@@ -200,7 +201,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 					bp.VirtualNetwork = nil
 					if err := bc.CreateOrUpdateLBBackendPool(lbName, bp); err != nil {
 						klog.Errorf("bc.ReconcileBackendPools for service (%s): failed to cleanup IP based backend pool %s: %s", serviceName, lbBackendPoolNames[isIPv6], err.Error())
-						return false, false, false, fmt.Errorf("bc.ReconcileBackendPools for service (%s): failed to cleanup IP based backend pool %s: %w", serviceName, lbBackendPoolNames[isIPv6], err)
+						return false, false, fmt.Errorf("bc.ReconcileBackendPools for service (%s): failed to cleanup IP based backend pool %s: %w", serviceName, lbBackendPoolNames[isIPv6], err)
 					}
 					newBackendPools[i] = bp
 					lb.BackendAddressPools = &newBackendPools
@@ -218,7 +219,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 							klog.V(2).Infof("bc.ReconcileBackendPools for service (%s): vm not found for ipConfID %s", serviceName, ipConfID)
 							bipConfigNotFound = append(bipConfigNotFound, ipConf)
 						} else {
-							return false, false, false, err
+							return false, false, err
 						}
 					}
 
@@ -229,7 +230,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 					shouldExcludeLoadBalancer, err := bc.ShouldNodeExcludedFromLoadBalancer(nodeName)
 					if err != nil {
 						klog.Errorf("bc.ReconcileBackendPools: ShouldNodeExcludedFromLoadBalancer(%s) failed with error: %v", nodeName, err)
-						return false, false, false, err
+						return false, false, err
 					}
 					if shouldExcludeLoadBalancer {
 						klog.V(2).Infof("bc.ReconcileBackendPools for service (%s): lb backendpool - found unwanted node %s, decouple it from the LB %s", serviceName, nodeName, lbName)
@@ -257,7 +258,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 		// decouple the backendPool from the node
 		updated, err := bc.VMSet.EnsureBackendPoolDeleted(service, lbBackendPoolIDsSlice, vmSetName, &backendpoolToBeDeleted, false)
 		if err != nil {
-			return false, false, false, err
+			return false, false, err
 		}
 		if updated {
 			shouldRefreshLB = true
@@ -267,7 +268,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 	if shouldRefreshLB {
 		lb, _, err = bc.getAzureLoadBalancer(lbName, cache.CacheReadTypeForceRefresh)
 		if err != nil {
-			return false, false, false, fmt.Errorf("bc.ReconcileBackendPools for service (%s): failed to get loadbalancer %s: %w", serviceName, lbName, err)
+			return false, false, fmt.Errorf("bc.ReconcileBackendPools for service (%s): failed to get loadbalancer %s: %w", serviceName, lbName, err)
 		}
 	}
 
@@ -288,7 +289,7 @@ func (bc *backendPoolTypeNodeIPConfig) ReconcileBackendPools(clusterName string,
 	}
 
 	isOperationSucceeded = true
-	return isBackendPoolPreConfigured, changed, false, err
+	return isBackendPoolPreConfigured, changed, err
 }
 
 func getBackendIPConfigurationsToBeDeleted(
@@ -420,6 +421,33 @@ func (bi *backendPoolTypeNodeIP) EnsureHostsInPool(service *v1.Service, nodes []
 				continue
 			}
 
+			var err error
+			shouldSkip := false
+			useSingleSLB := strings.EqualFold(bi.LoadBalancerSku, consts.LoadBalancerSkuStandard) && !bi.EnableMultipleStandardLoadBalancers
+			if !useSingleSLB {
+				vmSetName, err = bi.VMSet.GetNodeVMSetName(node)
+				if err != nil {
+					klog.Errorf("bi.EnsureHostsInPool: failed to get vmSet name by node name: %s", err.Error())
+					return err
+				}
+
+				if !strings.EqualFold(vmSetName, bi.mapLoadBalancerNameToVMSet(lbName, clusterName)) {
+					shouldSkip = true
+
+					lbNamePrefix := strings.TrimSuffix(lbName, consts.InternalLoadBalancerNameSuffix)
+					if strings.EqualFold(lbNamePrefix, clusterName) &&
+						strings.EqualFold(bi.LoadBalancerSku, consts.LoadBalancerSkuStandard) &&
+						bi.getVMSetNamesSharingPrimarySLB().Has(vmSetName) {
+						klog.V(4).Infof("bi.EnsureHostsInPool: the node %s in VMSet %s is supposed to share the primary SLB", node.Name, vmSetName)
+						shouldSkip = false
+					}
+				}
+			}
+			if shouldSkip {
+				klog.V(4).Infof("bi.EnsureHostsInPool: skipping attaching node %s to lb %s, because the vmSet of the node is %s", node.Name, lbName, vmSetName)
+				continue
+			}
+
 			privateIP := getNodePrivateIPAddress(node, isIPv6)
 			if !existingIPs.Has(privateIP) {
 				name := node.Name
@@ -508,7 +536,7 @@ func (bi *backendPoolTypeNodeIP) CleanupVMSetFromBackendPoolByCondition(slb *net
 	return slb, nil
 }
 
-func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, bool, error) {
+func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, error) {
 	var newBackendPools []network.BackendAddressPool
 	if lb.BackendAddressPools != nil {
 		newBackendPools = *lb.BackendAddressPools
@@ -524,32 +552,17 @@ func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, servi
 	isBackendPoolPreConfigured := bi.isBackendPoolPreConfigured(service)
 
 	mc := metrics.NewMetricContext("services", "migrate_to_ip_based_backend_pool", bi.ResourceGroup, bi.getNetworkResourceSubscriptionID(), serviceName)
-
-	var (
-		err                   error
-		bpIdxes               []int
-		lbBackendPoolIDsSlice []string
-	)
-	nicsCountMap := make(map[string]int)
+	var err error
+	bpIdxes := []int{}
+	lbBackendPoolIDsSlice := []string{}
 	for i := len(newBackendPools) - 1; i >= 0; i-- {
 		bp := newBackendPools[i]
 		found, isIPv6 := isLBBackendPoolsExisting(lbBackendPoolNames, bp.Name)
 		if found {
 			bpIdxes = append(bpIdxes, i)
-			klog.V(10).Infof("bi.ReconcileBackendPools for service (%s): found wanted backendpool. Not adding anything", serviceName)
+			klog.V(10).Infof("bi.ReconcileBackendPools for service (%s): found wanted backendpool. not adding anything", serviceName)
 			foundBackendPools[isIPv6] = true
 			lbBackendPoolIDsSlice = append(lbBackendPoolIDsSlice, lbBackendPoolIDs[isIPv6])
-
-			if nicsCount := countNICsOnBackendPool(bp); nicsCount > 0 {
-				nicsCountMap[pointer.StringDeref(bp.Name, "")] = nicsCount
-				klog.V(4).Infof(
-					"bi.ReconcileBackendPools for service(%s): found NIC-based backendpool %s with %d NICs, will migrate to IP-based",
-					serviceName,
-					pointer.StringDeref(bp.Name, ""),
-					nicsCount,
-				)
-				isMigration = true
-			}
 		} else {
 			klog.V(10).Infof("bi.ReconcileBackendPools for service (%s): found unmanaged backendpool %s", serviceName, *bp.Name)
 		}
@@ -560,45 +573,14 @@ func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, servi
 		// If the LB backend pool type is configured from nodeIPConfiguration
 		// to nodeIP, we need to decouple the VM NICs from the LB
 		// before attaching nodeIPs/podIPs to the LB backend pool.
-		// If the migration API is enabled, we use the migration API to decouple
-		// the VM NICs from the LB. Then we manually decouple the VMSS
-		// and its VMs from the LB by EnsureBackendPoolDeleted. These manual operations
-		// cannot be omitted because we use the VMSS manual upgrade policy.
-		// If the migration API is not enabled, we manually decouple the VM NICs and
-		// the VMSS from the LB by EnsureBackendPoolDeleted. If no NIC-based backend
-		// pool is found (it is not a migration scenario), EnsureBackendPoolDeleted would be a no-op.
-		if isMigration && bi.EnableMigrateToIPBasedBackendPoolAPI {
-			var backendPoolNames []string
-			for _, id := range lbBackendPoolIDsSlice {
-				name, err := getLBNameFromBackendPoolID(id)
-				if err != nil {
-					klog.Errorf("bi.ReconcileBackendPools for service (%s): failed to get LB name from backend pool ID: %s", serviceName, err.Error())
-					return false, false, false, err
-				}
-				backendPoolNames = append(backendPoolNames, name)
-			}
-
-			if err := bi.MigrateToIPBasedBackendPoolAndWaitForCompletion(lbName, backendPoolNames, nicsCountMap); err != nil {
-				backendPoolNamesStr := strings.Join(backendPoolNames, ",")
-				klog.Errorf("Failed to migrate to IP based backend pool for lb %s, backend pool %s: %s", lbName, backendPoolNamesStr, err.Error())
-				return false, false, false, err
-			}
-		}
-
-		// EnsureBackendPoolDeleted is useful in the following scenarios:
-		// 1. Migrate from NIC-based to IP-based backend pool if the migration
-		// API is not enabled.
-		// 2. Migrate from NIC-based to IP-based backend pool when the migration
-		// API is enabled. This is needed because since we use the manual upgrade
-		// policy on VMSS so the migration API will not change the VMSS and VMSS
-		// VMs during the migration.
-		// 3. Decouple vmss from the lb if the backend pool is empty when using
-		// ip-based LB. Ref: https://github.com/kubernetes-sigs/cloud-provider-azure/pull/2829.
 		klog.V(2).Infof("bi.ReconcileBackendPools for service (%s) and vmSet (%s): ensuring the LB is decoupled from the VMSet", serviceName, vmSetName)
 		shouldRefreshLB, err = bi.VMSet.EnsureBackendPoolDeleted(service, lbBackendPoolIDsSlice, vmSetName, lb.BackendAddressPools, true)
 		if err != nil {
 			klog.Errorf("bi.ReconcileBackendPools for service (%s): failed to EnsureBackendPoolDeleted: %s", serviceName, err.Error())
-			return false, false, false, err
+			return false, false, err
+		}
+		if shouldRefreshLB {
+			isMigration = true
 		}
 
 		for _, i := range bpIdxes {
@@ -643,14 +625,19 @@ func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, servi
 			if updated {
 				(*lb.BackendAddressPools)[i] = bp
 				if err := bi.CreateOrUpdateLBBackendPool(lbName, bp); err != nil {
-					return false, false, false, fmt.Errorf("bi.ReconcileBackendPools for service (%s): lb backendpool - failed to update backend pool %s for load balancer %s: %w", serviceName, pointer.StringDeref(bp.Name, ""), lbName, err)
+					return false, false, fmt.Errorf("bi.ReconcileBackendPools for service (%s): lb backendpool - failed to update backend pool %s for load balancer %s: %w", serviceName, pointer.StringDeref(bp.Name, ""), lbName, err)
 				}
 				shouldRefreshLB = true
 			}
 		}
 	}
 
-	shouldRefreshLB = shouldRefreshLB || isMigration
+	if shouldRefreshLB {
+		lb, _, err = bi.getAzureLoadBalancer(lbName, cache.CacheReadTypeForceRefresh)
+		if err != nil {
+			return false, false, fmt.Errorf("bi.ReconcileBackendPools for service (%s): failed to get load balancer %s: %w", serviceName, lbName, err)
+		}
+	}
 
 	for _, ipFamily := range service.Spec.IPFamilies {
 		if foundBackendPools[ipFamily == v1.IPv6Protocol] {
@@ -669,7 +656,7 @@ func (bi *backendPoolTypeNodeIP) ReconcileBackendPools(clusterName string, servi
 	}
 
 	isOperationSucceeded = true
-	return isBackendPoolPreConfigured, changed, shouldRefreshLB, nil
+	return isBackendPoolPreConfigured, changed, nil
 }
 
 func (bi *backendPoolTypeNodeIP) GetBackendPrivateIPs(clusterName string, service *v1.Service, lb *network.LoadBalancer) ([]string, []string) {
diff --git a/pkg/provider/azure_mock_loadbalancer_backendpool.go b/pkg/provider/azure_mock_loadbalancer_backendpool.go
index 9d1911995..474eac6be 100644
--- a/pkg/provider/azure_mock_loadbalancer_backendpool.go
+++ b/pkg/provider/azure_mock_loadbalancer_backendpool.go
@@ -21,33 +21,48 @@ import (
 
 	network "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2022-07-01/network"
 	gomock "github.com/golang/mock/gomock"
+
 	v1 "k8s.io/api/core/v1"
 )
 
-// MockBackendPool is a mock of BackendPool interface.
+// MockBackendPool is a mock of BackendPool interface
 type MockBackendPool struct {
 	ctrl     *gomock.Controller
 	recorder *MockBackendPoolMockRecorder
 }
 
-// MockBackendPoolMockRecorder is the mock recorder for MockBackendPool.
+// MockBackendPoolMockRecorder is the mock recorder for MockBackendPool
 type MockBackendPoolMockRecorder struct {
 	mock *MockBackendPool
 }
 
-// NewMockBackendPool creates a new mock instance.
+// NewMockBackendPool creates a new mock instance
 func NewMockBackendPool(ctrl *gomock.Controller) *MockBackendPool {
 	mock := &MockBackendPool{ctrl: ctrl}
 	mock.recorder = &MockBackendPoolMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use.
+// EXPECT returns an object that allows the caller to indicate expected use
 func (m *MockBackendPool) EXPECT() *MockBackendPoolMockRecorder {
 	return m.recorder
 }
 
-// CleanupVMSetFromBackendPoolByCondition mocks base method.
+// EnsureHostsInPool mocks base method
+func (m *MockBackendPool) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, backendPoolID, vmSetName, clusterName, lbName string, backendPool network.BackendAddressPool) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "EnsureHostsInPool", service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// EnsureHostsInPool indicates an expected call of EnsureHostsInPool
+func (mr *MockBackendPoolMockRecorder) EnsureHostsInPool(service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EnsureHostsInPool", reflect.TypeOf((*MockBackendPool)(nil).EnsureHostsInPool), service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool)
+}
+
+// CleanupVMSetFromBackendPoolByCondition mocks base method
 func (m *MockBackendPool) CleanupVMSetFromBackendPoolByCondition(slb *network.LoadBalancer, service *v1.Service, nodes []*v1.Node, clusterName string, shouldRemoveVMSetFromSLB func(string) bool) (*network.LoadBalancer, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "CleanupVMSetFromBackendPoolByCondition", slb, service, nodes, clusterName, shouldRemoveVMSetFromSLB)
@@ -56,27 +71,29 @@ func (m *MockBackendPool) CleanupVMSetFromBackendPoolByCondition(slb *network.Lo
 	return ret0, ret1
 }
 
-// CleanupVMSetFromBackendPoolByCondition indicates an expected call of CleanupVMSetFromBackendPoolByCondition.
+// CleanupVMSetFromBackendPoolByCondition indicates an expected call of CleanupVMSetFromBackendPoolByCondition
 func (mr *MockBackendPoolMockRecorder) CleanupVMSetFromBackendPoolByCondition(slb, service, nodes, clusterName, shouldRemoveVMSetFromSLB interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CleanupVMSetFromBackendPoolByCondition", reflect.TypeOf((*MockBackendPool)(nil).CleanupVMSetFromBackendPoolByCondition), slb, service, nodes, clusterName, shouldRemoveVMSetFromSLB)
 }
 
-// EnsureHostsInPool mocks base method.
-func (m *MockBackendPool) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node, backendPoolID, vmSetName, clusterName, lbName string, backendPool network.BackendAddressPool) error {
+// ReconcileBackendPools mocks base method
+func (m *MockBackendPool) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "EnsureHostsInPool", service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool)
-	ret0, _ := ret[0].(error)
-	return ret0
+	ret := m.ctrl.Call(m, "ReconcileBackendPools", clusterName, service, lb)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(bool)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
 }
 
-// EnsureHostsInPool indicates an expected call of EnsureHostsInPool.
-func (mr *MockBackendPoolMockRecorder) EnsureHostsInPool(service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool interface{}) *gomock.Call {
+// ReconcileBackendPools indicates an expected call of ReconcileBackendPools
+func (mr *MockBackendPoolMockRecorder) ReconcileBackendPools(clusterName, service, lb interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EnsureHostsInPool", reflect.TypeOf((*MockBackendPool)(nil).EnsureHostsInPool), service, nodes, backendPoolID, vmSetName, clusterName, lbName, backendPool)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconcileBackendPools", reflect.TypeOf((*MockBackendPool)(nil).ReconcileBackendPools), clusterName, service, lb)
 }
 
-// GetBackendPrivateIPs mocks base method.
+// GetBackendPrivateIPs mocks base method
 func (m *MockBackendPool) GetBackendPrivateIPs(clusterName string, service *v1.Service, lb *network.LoadBalancer) ([]string, []string) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetBackendPrivateIPs", clusterName, service, lb)
@@ -85,25 +102,8 @@ func (m *MockBackendPool) GetBackendPrivateIPs(clusterName string, service *v1.S
 	return ret0, ret1
 }
 
-// GetBackendPrivateIPs indicates an expected call of GetBackendPrivateIPs.
+// GetBackendPrivateIPs indicates an expected call of GetBackendPrivateIPs
 func (mr *MockBackendPoolMockRecorder) GetBackendPrivateIPs(clusterName, service, lb interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBackendPrivateIPs", reflect.TypeOf((*MockBackendPool)(nil).GetBackendPrivateIPs), clusterName, service, lb)
 }
-
-// ReconcileBackendPools mocks base method.
-func (m *MockBackendPool) ReconcileBackendPools(clusterName string, service *v1.Service, lb *network.LoadBalancer) (bool, bool, bool, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ReconcileBackendPools", clusterName, service, lb)
-	ret0, _ := ret[0].(bool)
-	ret1, _ := ret[1].(bool)
-	ret2, _ := ret[2].(bool)
-	ret3, _ := ret[3].(error)
-	return ret0, ret1, ret2, ret3
-}
-
-// ReconcileBackendPools indicates an expected call of ReconcileBackendPools.
-func (mr *MockBackendPoolMockRecorder) ReconcileBackendPools(clusterName, service, lb interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconcileBackendPools", reflect.TypeOf((*MockBackendPool)(nil).ReconcileBackendPools), clusterName, service, lb)
-}
diff --git a/pkg/provider/azure_mock_vmsets.go b/pkg/provider/azure_mock_vmsets.go
index c4102e2e9..2fedad0ab 100644
--- a/pkg/provider/azure_mock_vmsets.go
+++ b/pkg/provider/azure_mock_vmsets.go
@@ -24,7 +24,6 @@ import (
 	network "github.com/Azure/azure-sdk-for-go/services/network/mgmt/2022-07-01/network"
 	azure "github.com/Azure/go-autorest/autorest/azure"
 	gomock "github.com/golang/mock/gomock"
-
 	v1 "k8s.io/api/core/v1"
 	types "k8s.io/apimachinery/pkg/types"
 	cloud_provider "k8s.io/cloud-provider"
@@ -416,7 +415,7 @@ func (mr *MockVMSetMockRecorder) UpdateVM(ctx, nodeName interface{}) *gomock.Cal
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVM", reflect.TypeOf((*MockVMSet)(nil).UpdateVM), ctx, nodeName)
 }
 
-// UpdateVMAsync mocks base method.
+// UpdateVMAsync mocks base method
 func (m *MockVMSet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName) (*azure.Future, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "UpdateVMAsync", ctx, nodeName)
@@ -425,7 +424,7 @@ func (m *MockVMSet) UpdateVMAsync(ctx context.Context, nodeName types.NodeName)
 	return ret0, ret1
 }
 
-// UpdateVMAsync indicates an expected call of UpdateVMAsync.
+// UpdateVMAsync indicates an expected call of UpdateVMAsync
 func (mr *MockVMSetMockRecorder) UpdateVMAsync(ctx, nodeName interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateVMAsync", reflect.TypeOf((*MockVMSet)(nil).UpdateVMAsync), ctx, nodeName)
@@ -434,7 +433,7 @@ func (mr *MockVMSetMockRecorder) UpdateVMAsync(ctx, nodeName interface{}) *gomoc
 // WaitForUpdateResult mocks base method.
 func (m *MockVMSet) WaitForUpdateResult(ctx context.Context, future *azure.Future, nodeName types.NodeName, source string) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "WaitForUpdateResult", ctx, future, nodeName, source)
+	ret := m.ctrl.Call(m, "WaitForUpdateResult", ctx, future, source)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
@@ -442,5 +441,5 @@ func (m *MockVMSet) WaitForUpdateResult(ctx context.Context, future *azure.Futur
 // WaitForUpdateResult indicates an expected call of WaitForUpdateResult.
 func (mr *MockVMSetMockRecorder) WaitForUpdateResult(ctx, future, nodeName, source interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WaitForUpdateResult", reflect.TypeOf((*MockVMSet)(nil).WaitForUpdateResult), ctx, future, nodeName, source)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WaitForUpdateResult", reflect.TypeOf((*MockVMSet)(nil).WaitForUpdateResult), ctx, future, source)
 }
diff --git a/pkg/provider/azure_standard.go b/pkg/provider/azure_standard.go
index c516a246a..519a46506 100644
--- a/pkg/provider/azure_standard.go
+++ b/pkg/provider/azure_standard.go
@@ -150,10 +150,15 @@ func (az *Cloud) getAzureLoadBalancerName(clusterName string, vmSetName string,
 	// The LB name prefix is set to the name of the cluster when:
 	// 1. the LB belongs to the primary agent pool.
 	// 2. using the single SLB.
-	if strings.EqualFold(vmSetName, az.VMSet.GetPrimaryVMSetName()) || az.useStandardLoadBalancer() {
+	useSingleSLB := az.useStandardLoadBalancer() && !az.EnableMultipleStandardLoadBalancers
+	if strings.EqualFold(vmSetName, az.VMSet.GetPrimaryVMSetName()) || useSingleSLB {
+		lbNamePrefix = clusterName
+	}
+	// 3. using multiple SLBs while the vmSet is sharing the primary SLB
+	useMultipleSLB := az.useStandardLoadBalancer() && az.EnableMultipleStandardLoadBalancers
+	if useMultipleSLB && az.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vmSetName)) {
 		lbNamePrefix = clusterName
 	}
-
 	if isInternal {
 		return fmt.Sprintf("%s%s", lbNamePrefix, consts.InternalLoadBalancerNameSuffix)
 	}
@@ -789,7 +794,8 @@ func (as *availabilitySet) getAgentPoolAvailabilitySets(vms []compute.VirtualMac
 // annotation would be ignored when using one SLB per cluster.
 func (as *availabilitySet) GetVMSetNames(service *v1.Service, nodes []*v1.Node) (availabilitySetNames *[]string, err error) {
 	hasMode, isAuto, serviceAvailabilitySetName := as.getServiceLoadBalancerMode(service)
-	if !hasMode || as.useStandardLoadBalancer() {
+	useSingleSLB := as.useStandardLoadBalancer() && !as.EnableMultipleStandardLoadBalancers
+	if !hasMode || useSingleSLB {
 		// no mode specified in service annotation or use single SLB mode
 		// default to PrimaryAvailabilitySetName
 		availabilitySetNames = &[]string{as.Config.PrimaryAvailabilitySetName}
@@ -923,6 +929,19 @@ func (as *availabilitySet) getPrimaryInterfaceWithVMSet(nodeName, vmSetName stri
 	if !as.useStandardLoadBalancer() {
 		// need to check the vmSet name when using the basic LB
 		needCheck = true
+	} else if as.EnableMultipleStandardLoadBalancers {
+		// need to check the vmSet name when using multiple standard LBs
+		needCheck = true
+
+		// ensure the vm that is supposed to share the primary SLB in the backendpool of the primary SLB
+		if machine.AvailabilitySet != nil {
+			vmasName, _ := getLastSegment(pointer.StringDeref(machine.AvailabilitySet.ID, ""), "/")
+			if strings.EqualFold(as.GetPrimaryVMSetName(), vmSetName) &&
+				as.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vmasName)) {
+				klog.V(4).Infof("getPrimaryInterfaceWithVMSet: the vm %s in the vmSet %s is supposed to share the primary SLB", nodeName, vmasName)
+				needCheck = false
+			}
+		}
 	}
 	if vmSetName != "" && needCheck {
 		expectedAvailabilitySetID := as.getAvailabilitySetID(nodeResourceGroup, vmSetName)
diff --git a/pkg/provider/azure_utils.go b/pkg/provider/azure_utils.go
index 35b660d0b..7637c4094 100644
--- a/pkg/provider/azure_utils.go
+++ b/pkg/provider/azure_utils.go
@@ -26,6 +26,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2022-07-01/network"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
 	"k8s.io/utils/pointer"
@@ -182,6 +183,18 @@ func (az *Cloud) reconcileTags(currentTagsOnResource, newTags map[string]*string
 	return currentTagsOnResource, changed
 }
 
+func (az *Cloud) getVMSetNamesSharingPrimarySLB() sets.Set[string] {
+	vmSetNames := make([]string, 0)
+	if az.NodePoolsWithoutDedicatedSLB != "" {
+		vmSetNames = strings.Split(az.Config.NodePoolsWithoutDedicatedSLB, consts.VMSetNamesSharingPrimarySLBDelimiter)
+		for i := 0; i < len(vmSetNames); i++ {
+			vmSetNames[i] = strings.ToLower(strings.TrimSpace(vmSetNames[i]))
+		}
+	}
+
+	return sets.New(vmSetNames...)
+}
+
 func getExtendedLocationTypeFromString(extendedLocationType string) network.ExtendedLocationTypes {
 	extendedLocationType = strings.ToLower(extendedLocationType)
 	if val, ok := strToExtendedLocationType[extendedLocationType]; ok {
@@ -507,38 +520,3 @@ func (az *Cloud) fillSubnet(subnet *network.Subnet, subnetName string) error {
 	}
 	return nil
 }
-
-func getLBNameFromBackendPoolID(backendPoolID string) (string, error) {
-	matches := backendPoolIDRE.FindStringSubmatch(backendPoolID)
-	if len(matches) != 2 {
-		return "", fmt.Errorf("backendPoolID %q is in wrong format", backendPoolID)
-	}
-
-	return matches[1], nil
-}
-
-func countNICsOnBackendPool(backendPool network.BackendAddressPool) int {
-	if backendPool.BackendAddressPoolPropertiesFormat == nil ||
-		backendPool.BackendIPConfigurations == nil {
-		return 0
-	}
-
-	return len(*backendPool.BackendIPConfigurations)
-}
-
-func countIPsOnBackendPool(backendPool network.BackendAddressPool) int {
-	if backendPool.BackendAddressPoolPropertiesFormat == nil ||
-		backendPool.LoadBalancerBackendAddresses == nil {
-		return 0
-	}
-
-	var ipsCount int
-	for _, loadBalancerBackendAddress := range *backendPool.LoadBalancerBackendAddresses {
-		if loadBalancerBackendAddress.LoadBalancerBackendAddressPropertiesFormat != nil &&
-			pointer.StringDeref(loadBalancerBackendAddress.IPAddress, "") != "" {
-			ipsCount++
-		}
-	}
-
-	return ipsCount
-}
diff --git a/pkg/provider/azure_vmss.go b/pkg/provider/azure_vmss.go
index c45bac687..2f7110f97 100644
--- a/pkg/provider/azure_vmss.go
+++ b/pkg/provider/azure_vmss.go
@@ -901,7 +901,8 @@ func (ss *ScaleSet) getAgentPoolScaleSets(nodes []*v1.Node) (*[]string, error) {
 // for loadbalancer exists then return the eligible VMSet.
 func (ss *ScaleSet) GetVMSetNames(service *v1.Service, nodes []*v1.Node) (*[]string, error) {
 	hasMode, isAuto, serviceVMSetName := ss.getServiceLoadBalancerMode(service)
-	if !hasMode || ss.useStandardLoadBalancer() {
+	useSingleSLB := ss.useStandardLoadBalancer() && !ss.EnableMultipleStandardLoadBalancers
+	if !hasMode || useSingleSLB {
 		// no mode specified in service annotation or use single SLB mode
 		// default to PrimaryScaleSetName
 		scaleSetNames := &[]string{ss.Config.PrimaryScaleSetName}
@@ -1092,6 +1093,17 @@ func (ss *ScaleSet) EnsureHostInPool(service *v1.Service, nodeName types.NodeNam
 	if !ss.useStandardLoadBalancer() {
 		// need to check the vmSet name when using the basic LB
 		needCheck = true
+	} else if ss.EnableMultipleStandardLoadBalancers {
+		// need to check the vmSet name when using multiple standard LBs
+		needCheck = true
+
+		// ensure the vm that is supposed to share the primary SLB in the backendpool of the primary SLB
+		if strings.EqualFold(ss.GetPrimaryVMSetName(), vmSetNameOfLB) &&
+			ss.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vm.VMSSName)) {
+			klog.V(4).Infof("EnsureHostInPool: the vm %s in the vmSet %s is supposed to share the primary SLB",
+				nodeName, vm.VMSSName)
+			needCheck = false
+		}
 	}
 
 	if vmSetNameOfLB != "" && needCheck && !strings.EqualFold(vmSetNameOfLB, vm.VMSSName) {
@@ -1204,7 +1216,7 @@ func (ss *ScaleSet) ensureVMSSInPool(service *v1.Service, nodes []*v1.Node, back
 
 	// the single standard load balancer supports multiple vmss in its backend while
 	// multiple standard load balancers and the basic load balancer doesn't
-	if ss.useStandardLoadBalancer() {
+	if ss.useStandardLoadBalancer() && !ss.EnableMultipleStandardLoadBalancers {
 		for _, node := range nodes {
 			if ss.excludeMasterNodesFromStandardLB() && isControlPlaneNode(node) {
 				continue
@@ -1543,7 +1555,7 @@ func (ss *ScaleSet) ensureBackendPoolDeletedFromNode(nodeName string, backendPoo
 
 	// Find primary network interface configuration.
 	if vm.VirtualMachineScaleSetVMProperties.NetworkProfileConfiguration.NetworkInterfaceConfigurations == nil {
-		klog.V(4).Infof("ensureBackendPoolDeletedFromNode: cannot obtain the primary network interface configuration, of vm %s, "+
+		klog.V(4).Infof("EnsureHostInPool: cannot obtain the primary network interface configuration, of vm %s, "+
 			"probably because the vm's being deleted", nodeName)
 		return "", "", "", nil, nil
 	}
@@ -1699,7 +1711,7 @@ func (ss *ScaleSet) ensureBackendPoolDeletedFromVmssUniform(backendPoolIDs []str
 
 	vmssNamesMap := make(map[string]bool)
 	// the standard load balancer supports multiple vmss in its backend while the basic sku doesn't
-	if ss.useStandardLoadBalancer() {
+	if ss.useStandardLoadBalancer() && !ss.EnableMultipleStandardLoadBalancers {
 		cachedUniform, err := ss.vmssCache.Get(consts.VMSSKey, azcache.CacheReadTypeDefault)
 		if err != nil {
 			klog.Errorf("ensureBackendPoolDeletedFromVMSS: failed to get vmss uniform from cache: %v", err)
diff --git a/pkg/provider/azure_vmssflex.go b/pkg/provider/azure_vmssflex.go
index 304238a51..b1d24e360 100644
--- a/pkg/provider/azure_vmssflex.go
+++ b/pkg/provider/azure_vmssflex.go
@@ -125,7 +125,8 @@ func (fs *FlexScaleSet) GetAgentPoolVMSetNames(nodes []*v1.Node) (*[]string, err
 // annotation would be ignored when using one SLB per cluster.
 func (fs *FlexScaleSet) GetVMSetNames(service *v1.Service, nodes []*v1.Node) (*[]string, error) {
 	hasMode, isAuto, serviceVMSetName := fs.getServiceLoadBalancerMode(service)
-	if !hasMode || fs.useStandardLoadBalancer() {
+	useSingleSLB := fs.useStandardLoadBalancer() && !fs.EnableMultipleStandardLoadBalancers
+	if !hasMode || useSingleSLB {
 		// no mode specified in service annotation or use single SLB mode
 		// default to PrimaryScaleSetName
 		vmssFlexNames := &[]string{fs.Config.PrimaryScaleSetName}
@@ -480,6 +481,18 @@ func (fs *FlexScaleSet) EnsureHostInPool(service *v1.Service, nodeName types.Nod
 	if !fs.useStandardLoadBalancer() {
 		return "", "", "", nil, fmt.Errorf("EnsureHostInPool: VMSS Flex does not support Basic Load Balancer")
 	}
+	if fs.EnableMultipleStandardLoadBalancers {
+		// need to check the vmSet name when using multiple standard LBs
+		needCheck = true
+
+		// ensure the vm that is supposed to share the primary SLB in the backendpool of the primary SLB
+		if strings.EqualFold(fs.GetPrimaryVMSetName(), vmSetNameOfLB) &&
+			fs.getVMSetNamesSharingPrimarySLB().Has(strings.ToLower(vmssFlexName)) {
+			klog.V(4).Infof("EnsureHostInPool: the vm %s in the vmSet %s is supposed to share the primary SLB",
+				nodeName, vmssFlexName)
+			needCheck = false
+		}
+	}
 	if vmSetNameOfLB != "" && needCheck && !strings.EqualFold(vmSetNameOfLB, vmssFlexName) {
 		klog.V(3).Infof("EnsureHostInPool skips node %s because it is not in the ScaleSet %s", name, vmSetNameOfLB)
 		return "", "", "", nil, errNotInVMSet
@@ -581,7 +594,7 @@ func (fs *FlexScaleSet) ensureVMSSFlexInPool(service *v1.Service, nodes []*v1.No
 
 	// the single standard load balancer supports multiple vmss in its backend while
 	// multiple standard load balancers doesn't
-	if fs.useStandardLoadBalancer() {
+	if fs.useStandardLoadBalancer() && !fs.EnableMultipleStandardLoadBalancers {
 		for _, node := range nodes {
 			if fs.excludeMasterNodesFromStandardLB() && isControlPlaneNode(node) {
 				continue
@@ -774,7 +787,7 @@ func (fs *FlexScaleSet) EnsureHostsInPool(service *v1.Service, nodes []*v1.Node,
 
 func (fs *FlexScaleSet) ensureBackendPoolDeletedFromVmssFlex(backendPoolIDs []string, vmSetName string) error {
 	vmssNamesMap := make(map[string]bool)
-	if fs.useStandardLoadBalancer() {
+	if fs.useStandardLoadBalancer() && !fs.EnableMultipleStandardLoadBalancers {
 		cached, err := fs.vmssFlexCache.Get(consts.VmssFlexKey, azcache.CacheReadTypeDefault)
 		if err != nil {
 			klog.Errorf("ensureBackendPoolDeletedFromVmssFlex: failed to get vmss flex from cache: %v", err)
@@ -923,7 +936,7 @@ func (fs *FlexScaleSet) EnsureBackendPoolDeleted(service *v1.Service, backendPoo
 		}
 		// only vmsses in the resource group same as it's in azure config are included
 		if strings.EqualFold(resourceGroupName, fs.ResourceGroup) {
-			if fs.useStandardLoadBalancer() {
+			if fs.useStandardLoadBalancer() && !fs.EnableMultipleStandardLoadBalancers {
 				vmssFlexVMNameMap[nodeName] = nicName
 			} else {
 				if strings.EqualFold(vmssFlexName, vmSetName) {
diff --git a/site/content/en/development/deploy-cluster.md b/site/content/en/development/deploy-cluster.md
index f2c382aab..f6feb76bb 100644
--- a/site/content/en/development/deploy-cluster.md
+++ b/site/content/en/development/deploy-cluster.md
@@ -40,6 +40,7 @@ Customizations are supported by environment variables:
 | AZURE_CLOUD_NODE_MANAGER_IMG            | false    | image of the cloud-node-manager                                                    | mcr.microsoft.com/oss/kubernetes/azure-cloud-node-manager:v1.23.1       |
 | KUBERNETES_VERSION                      | false    | Kubernetes components version                                                      | v1.25.0                                                                 |
 | AZURE_LOADBALANCER_SKU                  | false    | LoadBalancer SKU, Standard or Basic                                                | Standard                                                                |
+| ENABLE_MULTI_SLB                        | false    | Enable multiple standard LoadBalancers per cluster                                 | false                                                                   |
 | LB_BACKEND_POOL_CONFIG_TYPE             | false    | LoadBalancer backend pool configuration type, nodeIPConfiguration, nodeIP or podIP | nodeIPConfiguration                                                     |
 | PUT_VMSS_VM_BATCH_SIZE                  | false    | Batch size when updating VMSS VM concurrently                                      | 0                                                                       |
 | AZURE_SSH_PUBLIC_KEY                    | false    | SSH public key to connecet to the VMs                                              | ""                                                                      |
diff --git a/site/content/en/install/configs.md b/site/content/en/install/configs.md
index 0ef75aaac..0c57621fc 100644
--- a/site/content/en/install/configs.md
+++ b/site/content/en/install/configs.md
@@ -67,16 +67,16 @@ If more than one value is set, the order is `Managed Identity` > `Service Princi
 
 | Name                                                       | Description                                                                                                                                                                                                       | Remark                                                                                                                                |
 |------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------|
-| resourceGroup                                              | The name of the resource group that the cluster is deployed in                                                                                                                                                    |                                                                                                                                       |
-| location                                                   | The location of the resource group that the cluster is deployed in                                                                                                                                                |                                                                                                                                       |
-| vnetName                                                   | The name of the VNet that the cluster is deployed in                                                                                                                                                              |                                                                                                                                       |
-| vnetResourceGroup                                          | The name of the resource group that the Vnet is deployed in                                                                                                                                                       |                                                                                                                                       |
-| subnetName                                                 | The name of the subnet that the cluster is deployed in                                                                                                                                                            |                                                                                                                                       |
-| securityGroupName                                          | The name of the security group attached to the cluster's subnet                                                                                                                                                   |                                                                                                                                       |
-| securityGroupResourceGroup                                 | The name of the resource group that the security group is deployed in                                                                                                                                             |                                                                                                                                       |
+| resourceGroup                                              | The name of the resource group that the cluster is deployed in                                                                                                                                                    ||
+| location                                                   | The location of the resource group that the cluster is deployed in                                                                                                                                                ||
+| vnetName                                                   | The name of the VNet that the cluster is deployed in                                                                                                                                                              ||
+| vnetResourceGroup                                          | The name of the resource group that the Vnet is deployed in                                                                                                                                                       ||
+| subnetName                                                 | The name of the subnet that the cluster is deployed in                                                                                                                                                            ||
+| securityGroupName                                          | The name of the security group attached to the cluster's subnet                                                                                                                                                   ||
+| securityGroupResourceGroup                                 | The name of the resource group that the security group is deployed in                                                                                                                                             ||
 | routeTableName                                             | The name of the route table attached to the subnet that the cluster is deployed in                                                                                                                                | Optional in 1.6                                                                                                                       |
 | primaryAvailabilitySetName[*](#primaryavailabilitysetname) | The name of the availability set that should be used as the load balancer backend                                                                                                                                 | Optional                                                                                                                              |
-| vmType                                                     | The type of azure nodes. Candidate values are: `vmss`, `vmssflex` and `standard`                                                                                                                                  | Optional, default to `standard`                                                                                                       |
+| vmType                                                     | The type of azure nodes. Candidate values are: `vmss`, `vmssflex` and `standard`                                                                                                                                              | Optional, default to `standard`                                                                                                       |
 | primaryScaleSetName[*](#primaryscalesetname)               | The name of the scale set that should be used as the load balancer backend                                                                                                                                        | Optional                                                                                                                              |
 | cloudProviderBackoff                                       | Enable exponential backoff to manage resource request retries                                                                                                                                                     | Boolean value, default to false                                                                                                       |
 | cloudProviderBackoffRetries                                | Backoff retry limit                                                                                                                                                                                               | Integer value, valid if `cloudProviderBackoff` is true                                                                                |
@@ -97,7 +97,7 @@ If more than one value is set, the order is `Managed Identity` > `Service Princi
 | routeTableResourceGroup                                    | The resource group name for routeTable                                                                                                                                                                            | Default same as resourceGroup and available since v1.15.0                                                                             |
 | loadBalancerName                                           | Working together with loadBalancerResourceGroup to determine the LB name in a different resource group                                                                                                            | Since v1.18.0, default is cluster name setting on kube-controller-manager                                                             |
 | loadBalancerResourceGroup                                  | The load balancer resource group name, which is different from node resource group                                                                                                                                | Since v1.18.0, default is same as resourceGroup                                                                                       |
-| disableAvailabilitySetNodes                                | Disable supporting for AvailabilitySet virtual machines in vmss cluster. It should only be used when vmType is "vmss" and all the nodes (including master) are VMSS Uniform virtual machines                      | Since v1.18.0, default is false                                                                                                       |
+| disableAvailabilitySetNodes                                | Disable supporting for AvailabilitySet virtual machines in vmss cluster. It should only be used when vmType is "vmss" and all the nodes (including master) are VMSS Uniform virtual machines                      | Since v1.18.0, default is false      
 | enableVmssFlexNodes                                        | Enable supporting for VMSS Flex virtual machines in vmss cluster. It should only be used when vmType is "vmss"                                                                                                    | Since v1.26.0, default is false                                                                                                       |
 | availabilitySetNodesCacheTTLInSeconds                      | Cache TTL in seconds for availabilitySet Nodes                                                                                                                                                                    | Since v1.18.0, default is 900                                                                                                         |
 | vmssCacheTTLInSeconds                                      | Cache TTL in seconds for VMSS                                                                                                                                                                                     | Since v1.18.0, default is 600                                                                                                         |
@@ -110,9 +110,9 @@ If more than one value is set, the order is `Managed Identity` > `Service Princi
 | tags                                                       | Tags that would be tagged onto the cloud provider managed resources, including lb, public IP, network security group and route table.                                                                             | Optional. Supported since v1.20.0.                                                                                                    |
 | tagsMap                                                    | JSON-style tags, will be merged with `tags`                                                                                                                                                                       | Optional. Supported since v1.23.0.                                                                                                    |
 | systemTags                                                 | Tag keys that should not be deleted when being updated.                                                                                                                                                           | Optional. Supported since v1.21.0.                                                                                                    |
+| enableMultipleStandardLoadBalancers                        | Enable multiple standard Load Balancers per cluster.                                                                                                                                                              | Optional. Supported since v1.20.0                                                                                                     |
 | loadBalancerBackendPoolConfigurationType                   | The type of the Load Balancer backend pool. Supported values are `nodeIPConfiguration` (default) and `nodeIP`                                                                                                     | Optional. Supported since v1.23.0                                                                                                     |
 | putVMSSVMBatchSize                                         | The number of requests the client sends concurrently in a batch when putting the VMSS VMs. Anything smaller than or equal to 0 means to update VMSS VMs one by one in sequence.                                   | Optional. Supported since v1.24.0.                                                                                                    |
-| enableMigrateToIPBasedBackendPoolAPI                       | Use the migration API to migrate from NIC-based to IP-based Load Balancer backend pools without downtime.                                                                                                         | Optional. Supported since v1.24.0.                                                                                                    |
 
 ### primaryAvailabilitySetName
 
diff --git a/site/content/en/topics/assumptions.md b/site/content/en/topics/assumptions.md
index 556c39ed5..d327fd21e 100644
--- a/site/content/en/topics/assumptions.md
+++ b/site/content/en/topics/assumptions.md
@@ -13,6 +13,13 @@ Here is a list of Azure resource assumptions that are required for cloud provide
 
 * All Azure resources MUST be under the same tenant.
 * All virtual machine names MUST be the same as their hostname.
+* Node LoadBalancer's names SHOULD be following rules (`<clusterName>` is coming from `--cluster-name` configuration, default is `kubernetes`)
+  * When `enableMultipleStandardLoadBalancers` is configured to `false`, LoadBalancer's name SHOULD be `<clusterName>` for external type and `<clusterName>-internal` for internal type.
+  * When `enableMultipleStandardLoadBalancers` is configured to `true`, multiple standard load balancers SHOULD be provisioned:
+    * All the virtual machines MUST be part of either VirtualMachineScaleSet (VMSS) or AvailabilitySet (VMAS).
+    * Each VMAS and VMSS SHOULD be put behind a different standard LoadBalancer.
+    * The primary LoadBalancer's name SHOULD be `<clusterName>` for external type and `<clusterName>-internal` for internal type. Virtual machines that are part of primary VMAS (set by `primaryAvailabilitySetName`) or primary VMSS (set by `primaryScaleSetName`) SHOULD be added to primary LoadBalancer backend address pool.
+    * Other standard LoadBalancer's name SHOULD be same as VMAS or VMSS name.
 * The cluster name set for `kube-controller-manager --cluster-name=<cluster-name>` MUST not end with `-internal`.
 
 After the cluster is provisioned, cloud provider Azure MAY update the following Azure resources based on workloads:
diff --git a/site/content/en/topics/loadbalancer.md b/site/content/en/topics/loadbalancer.md
index de02bd06e..4bcf49a7b 100644
--- a/site/content/en/topics/loadbalancer.md
+++ b/site/content/en/topics/loadbalancer.md
@@ -151,6 +151,28 @@ spec:
       targetPort: 30102
 ```
 
+## Multiple Standard LoadBalancer per cluster
+
+> This feature is supported since v1.20.0.
+
+There is only one external and one internal Standard Load Balancer (SLB) at most per cluster. Set `enableMultipleStandardLoadBalancers=true` in the cloud config if you want to turn on the multiple SLB mode. Similar to the basic LB, there will be a 1:1 mapping between each SLB and VMSS/VMAS. The SLB of the primary VMSS/VMAS will be named after `clusterName` in the cloud config (in AKS, it would be `kubernetes`) while the name of those belonging to non-primary VMSS/VMAS will be the name of the corresponding vmSet.
+
+> If the cluster provisioning tools like ASK-Engine and CAPZ don't proactively create a dedicated SLB for each VMSS/VMAS when enabling multiple SLB, only the primary SLB would be created. You could manually trigger the creation by setting the service annotation `service.beta.kubernetes.io/azure-load-balancer-mode` to bind the service to that VMSS/VMAS. The dedicated SLB would be created once the service reconcile loop is done. Unlike the primary SLB, there is no default outbound rules/IPs for the non-primary SLBs. That means the SLB would be deleted once all the services referencing it are deleted.
+
+### Choose which SLB to use
+
+The service annotation `service.beta.kubernetes.io/azure-load-balancer-mode` will be respected as long as `enableMultipleStandardLoadBalancers=true` when using standard LB, and the usage is the same as it is in the basic LB clusters. Specifically, there are three selection mode: `default` to select the primary SLB; `__auto__` to select the SLB with minimum rules and `vmSetName` to select the dedicated SLB of that VMSS/VMAS.
+
+### Outbound Connections of non-primary VMSS/VMAS
+
+The outbound rules of the non-primary SLB are not managed by cloud provider azure. Instead, it should be managed by cluster provisioning tools. For now, there is no outbound configuration for the non-primary VMSS/VMAS, but we plan to support customized outbound configurations in AKS and CAPZ in the future.
+
+### Sharing the primary SLB with multiple VMSS/VMAS
+
+> This feature is supported since v1.21.0
+
+For each non-primary VMSS/VMAS, one can determine to use dedicated SLB or share the primary SLB. If the VMSS/VMAS names are in the cloud config `nodepoolsWithoutDedicatedSLB`, those would join the backend pool of the primary SLB while the others would remain to have dedicated SLBs. If the VMSS/VMAS supposed to share the primary SLB owns a dedicated SLB, the dedicated one would be deleted, and the VMSS/VMAS would be joint the primary SLB's backend pool.
+
 ## Custom Load Balancer health probe
 
 As documented [here](https://docs.microsoft.com/en-us/azure/load-balancer/load-balancer-custom-probe-overview), Tcp, Http and Https are three protocols supported by load balancer service.
@@ -272,16 +294,6 @@ The backend pool type can be configured by specifying `loadBalancerBackendPoolCo
 2. `nodeIP`. In this case we attach nodes to the LB by calling the LB API to add the node private IP addresses to the LB backend pool.
 3. `podIP` (not supported yet). In this case we do not attach nodes to the LB. Instead we directly adding pod IPs to the LB backend pool.
 
-To migrate from one backend pool type to another, just change the value of `loadBalancerBackendPoolConfigurationType` and re-apply the cloud configuration file. There will be downtime during the migration process.
-
-### Migration API from `nodeIPConfiguration` to `nodeIP`
-
-> This feature is supported since v1.24.0
-
-The migration from `nodeIPConfiguration` to `nodeIP` can be done without downtime by configuring `"enableMigrateToIPBasedBackendPoolAPI": true` in the cloud configuration file.
-
-```bash
-
 ## Load balancer limits
 
 The limits of the load balancer related resources are listed below:
@@ -313,3 +325,5 @@ The limit is up to 150 resources, in any combination of standalone virtual machi
 | Backend pool size                       | 300 IP configurations, single availability set |
 | Availability sets per Load Balancer     | 1                                              |
 | Load Balancers per VM                   | 2 (1 Public and 1 internal)                    |
+
+> There is a restriction of 300 rules per NIC, hence for single SLB mode 300 services are allowed at most. If more services are required, try to enable [multiple SLBs](../multiple-slb).


Test patch:
diff --git a/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient_test.go b/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient_test.go
index c33331a2e..9bc538f6e 100644
--- a/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient_test.go
+++ b/pkg/azureclients/loadbalancerclient/azure_loadbalancerclient_test.go
@@ -39,9 +39,8 @@ import (
 )
 
 const (
-	testResourceID            = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb1"
-	testBackendPoolResourceID = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb1/backendAddressPools/lb1"
-	testResourcePrefix        = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers"
+	testResourceID     = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb1"
+	testResourcePrefix = "/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers"
 )
 
 func TestNew(t *testing.T) {
@@ -329,67 +328,6 @@ func TestDelete(t *testing.T) {
 	}
 }
 
-func TestGetLBBackendPoolInternalError(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	response := &http.Response{
-		StatusCode: http.StatusInternalServerError,
-		Body:       io.NopCloser(bytes.NewReader([]byte("{}"))),
-	}
-	armClient := mockarmclient.NewMockInterface(ctrl)
-	armClient.EXPECT().GetResourceWithExpandQuery(gomock.Any(), testBackendPoolResourceID, "").Return(response, nil)
-	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any())
-
-	lbClient := getTestLoadBalancerClient(armClient)
-	expected := network.BackendAddressPool{Response: autorest.Response{}}
-	result, rerr := lbClient.GetLBBackendPool(context.TODO(), "rg", "lb1", "lb1", "")
-	assert.Equal(t, expected, result)
-	assert.NotNil(t, rerr)
-	assert.Equal(t, http.StatusInternalServerError, rerr.HTTPStatusCode)
-}
-
-func TestGetLBBackendPoolThrottle(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	response := &http.Response{
-		StatusCode: http.StatusTooManyRequests,
-		Body:       io.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(), testBackendPoolResourceID, "").Return(response, throttleErr)
-	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any())
-
-	lbClient := getTestLoadBalancerClient(armClient)
-	result, rerr := lbClient.GetLBBackendPool(context.TODO(), "rg", "lb1", "lb1", "")
-	assert.Empty(t, result)
-	assert.Equal(t, throttleErr, rerr)
-}
-
-func TestMigrateToIpBasedBackendPools(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	armClient := mockarmclient.NewMockInterface(ctrl)
-	response := &http.Response{
-		StatusCode: http.StatusOK,
-		Body:       io.NopCloser(bytes.NewReader([]byte(""))),
-	}
-	armClient.EXPECT().PostResource(gomock.Any(), testResourceID, "migrateToIpBased", backendPoolsToBeMigrated{BackendPoolNames: []string{"lb1"}}, gomock.Any()).Return(response, nil)
-	armClient.EXPECT().CloseResponse(gomock.Any(), gomock.Any())
-
-	lbClient := getTestLoadBalancerClient(armClient)
-	rerr := lbClient.MigrateToIPBasedBackendPool(context.TODO(), "rg", "lb1", []string{"lb1"})
-	assert.Nil(t, rerr)
-}
-
 func getTestLoadBalancer(name string) network.LoadBalancer {
 	return network.LoadBalancer{
 		ID:       pointer.String(fmt.Sprintf("/subscriptions/subscriptionID/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/%s", name)),
diff --git a/pkg/provider/azure_backoff_test.go b/pkg/provider/azure_backoff_test.go
index a4a098b95..05ca5ac43 100644
--- a/pkg/provider/azure_backoff_test.go
+++ b/pkg/provider/azure_backoff_test.go
@@ -666,83 +666,3 @@ func TestDeleteLBBackendPool(t *testing.T) {
 		assert.Equal(t, tc.expectedErr, err != nil)
 	}
 }
-
-func TestMigrateToIPBasedBackendPoolAndWaitForCompletion(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	for _, tc := range []struct {
-		desc                  string
-		migrationError        *retry.Error
-		backendPool           network.BackendAddressPool
-		backendPoolAfterRetry *network.BackendAddressPool
-		getBackendPoolError   *retry.Error
-		expectedError         error
-	}{
-		{
-			desc:           "MigrateToIPBasedBackendPoolAndWaitForCompletion should return the error if the migration fails",
-			migrationError: retry.NewError(false, errors.New("error")),
-			expectedError:  retry.NewError(false, errors.New("error")).Error(),
-		},
-		{
-			desc:                "MigrateToIPBasedBackendPoolAndWaitForCompletion should return the error if failed to get the backend pool",
-			getBackendPoolError: retry.NewError(false, errors.New("error")),
-			expectedError:       retry.NewError(false, errors.New("error")).Error(),
-		},
-		{
-			desc: "MigrateToIPBasedBackendPoolAndWaitForCompletion should retry if the number IPs on the backend pool is not expected",
-			backendPool: network.BackendAddressPool{
-				Name: pointer.String(testClusterName),
-				BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
-					LoadBalancerBackendAddresses: &[]network.LoadBalancerBackendAddress{
-						{
-							LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
-								IPAddress: pointer.String("1.2.3.4"),
-							},
-						},
-					},
-				},
-			},
-			backendPoolAfterRetry: &network.BackendAddressPool{
-				Name: pointer.String(testClusterName),
-				BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
-					LoadBalancerBackendAddresses: &[]network.LoadBalancerBackendAddress{
-						{
-							LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
-								IPAddress: pointer.String("1.2.3.4"),
-							},
-						},
-						{
-							LoadBalancerBackendAddressPropertiesFormat: &network.LoadBalancerBackendAddressPropertiesFormat{
-								IPAddress: pointer.String("2.3.4.5"),
-							},
-						},
-					},
-				},
-			},
-		},
-	} {
-		t.Run(tc.desc, func(t *testing.T) {
-			az := GetTestCloud(ctrl)
-			lbClient := mockloadbalancerclient.NewMockInterface(ctrl)
-			lbClient.EXPECT().MigrateToIPBasedBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(tc.migrationError)
-			if tc.migrationError == nil {
-				lbClient.EXPECT().GetLBBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(tc.backendPool, tc.getBackendPoolError)
-			}
-			if tc.backendPoolAfterRetry != nil {
-				lbClient.EXPECT().GetLBBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(*tc.backendPoolAfterRetry, nil)
-			}
-			az.LoadBalancerClient = lbClient
-
-			lbName := testClusterName
-			backendPoolNames := []string{testClusterName}
-			nicsCountMap := map[string]int{testClusterName: 2}
-			err := az.MigrateToIPBasedBackendPoolAndWaitForCompletion(lbName, backendPoolNames, nicsCountMap)
-			if tc.expectedError != nil {
-				assert.EqualError(t, err, tc.expectedError.Error())
-			} else {
-				assert.NoError(t, err)
-			}
-		})
-	}
-}
diff --git a/pkg/provider/azure_loadbalancer_backendpool_test.go b/pkg/provider/azure_loadbalancer_backendpool_test.go
index 25931d7f7..812c6ea55 100644
--- a/pkg/provider/azure_loadbalancer_backendpool_test.go
+++ b/pkg/provider/azure_loadbalancer_backendpool_test.go
@@ -44,6 +44,7 @@ func TestEnsureHostsInPoolNodeIP(t *testing.T) {
 
 	az := GetTestCloud(ctrl)
 	az.LoadBalancerSku = consts.LoadBalancerSkuStandard
+	az.EnableMultipleStandardLoadBalancers = true
 	bi := newBackendPoolTypeNodeIP(az)
 
 	nodes := []*v1.Node{
@@ -150,6 +151,8 @@ func TestEnsureHostsInPoolNodeIP(t *testing.T) {
 	for _, tc := range testcases {
 		t.Run(tc.desc, func(t *testing.T) {
 			mockVMSet := NewMockVMSet(ctrl)
+			mockVMSet.EXPECT().GetNodeVMSetName(gomock.Any()).Return("vmss-0", nil)
+			mockVMSet.EXPECT().GetPrimaryVMSetName().Return("vmss-0")
 			az.VMSet = mockVMSet
 
 			lbClient := mockloadbalancerclient.NewMockInterface(ctrl)
@@ -217,6 +220,7 @@ func TestCleanupVMSetFromBackendPoolByConditionNodeIPConfig(t *testing.T) {
 	defer ctrl.Finish()
 	cloud := GetTestCloud(ctrl)
 	cloud.LoadBalancerSku = consts.LoadBalancerSkuStandard
+	cloud.EnableMultipleStandardLoadBalancers = true
 	service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
 	lb := buildDefaultTestLB("testCluster", []string{
 		"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool1-00000000-nic-1/ipConfigurations/ipconfig1",
@@ -307,6 +311,7 @@ func TestCleanupVMSetFromBackendPoolForInstanceNotFound(t *testing.T) {
 	defer ctrl.Finish()
 	cloud := GetTestCloud(ctrl)
 	cloud.LoadBalancerSku = consts.LoadBalancerSkuStandard
+	cloud.EnableMultipleStandardLoadBalancers = true
 	cloud.PrimaryAvailabilitySetName = "agentpool1-availabilitySet-00000000"
 	clusterName := "testCluster"
 	service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
@@ -376,7 +381,7 @@ func TestReconcileBackendPoolsNodeIPConfig(t *testing.T) {
 
 	bc := newBackendPoolTypeNodeIPConfig(az)
 	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
-	_, _, _, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
+	_, _, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.NoError(t, err)
 
 	lb = network.LoadBalancer{
@@ -386,10 +391,9 @@ func TestReconcileBackendPoolsNodeIPConfig(t *testing.T) {
 	az = GetTestCloud(ctrl)
 	az.PreConfiguredBackendPoolLoadBalancerTypes = consts.PreConfiguredBackendPoolLoadBalancerTypesAll
 	bc = newBackendPoolTypeNodeIPConfig(az)
-	preConfigured, changed, shouldRefresh, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
+	preConfigured, changed, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.NoError(t, err)
 	assert.False(t, preConfigured)
-	assert.False(t, shouldRefresh)
 	assert.True(t, changed)
 }
 
@@ -417,11 +421,11 @@ func TestReconcileBackendPoolsNodeIPConfigRemoveIPConfig(t *testing.T) {
 
 	bc := newBackendPoolTypeNodeIPConfig(az)
 	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
-	_, _, _, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
+	_, _, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.NoError(t, err)
 
 	mockVMSet.EXPECT().GetNodeNameByIPConfigurationID("/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool1-00000000-nic-1/ipConfigurations/ipconfig1").Return("k8s-agentpool1-00000000", "", errors.New("error"))
-	_, _, _, err = bc.ReconcileBackendPools(testClusterName, &svc, &lb)
+	_, _, err = bc.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.Equal(t, "error", err.Error())
 }
 
@@ -445,7 +449,7 @@ func TestReconcileBackendPoolsNodeIPConfigPreConfigured(t *testing.T) {
 
 	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
 	bc := newBackendPoolTypeNodeIPConfig(az)
-	preConfigured, changed, _, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
+	preConfigured, changed, err := bc.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.True(t, preConfigured)
 	assert.False(t, changed)
 	assert.NoError(t, err)
@@ -465,14 +469,13 @@ func TestReconcileBackendPoolsNodeIPToIPConfig(t *testing.T) {
 
 	bc := newBackendPoolTypeNodeIPConfig(az)
 	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
-	_, _, _, err := bc.ReconcileBackendPools(testClusterName, &svc, lb)
+	_, _, err := bc.ReconcileBackendPools(testClusterName, &svc, lb)
 	assert.Contains(t, err.Error(), "create or update LB backend pool error")
 
 	lb = buildLBWithVMIPs(testClusterName, []string{"10.0.0.1", "10.0.0.2"})
 	mockLBClient.EXPECT().CreateOrUpdateBackendPools(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
-	_, _, shouldRefresh, err := bc.ReconcileBackendPools(testClusterName, &svc, lb)
+	_, _, err = bc.ReconcileBackendPools(testClusterName, &svc, lb)
 	assert.NoError(t, err)
-	assert.False(t, shouldRefresh)
 	assert.Empty(t, (*lb.BackendAddressPools)[0].LoadBalancerBackendAddresses)
 }
 
@@ -534,14 +537,14 @@ func TestReconcileBackendPoolsNodeIP(t *testing.T) {
 
 	lbClient := mockloadbalancerclient.NewMockInterface(ctrl)
 	lbClient.EXPECT().CreateOrUpdateBackendPools(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), bp, gomock.Any()).Return(nil)
+	lbClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(network.LoadBalancer{}, nil)
 	az.LoadBalancerClient = lbClient
 
 	bi := newBackendPoolTypeNodeIP(az)
 
 	service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
 
-	_, _, shouldRefresh, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
-	assert.True(t, shouldRefresh)
+	_, _, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
 	assert.NoError(t, err)
 
 	lb = &network.LoadBalancer{
@@ -551,10 +554,9 @@ func TestReconcileBackendPoolsNodeIP(t *testing.T) {
 	az = GetTestCloud(ctrl)
 	az.PreConfiguredBackendPoolLoadBalancerTypes = consts.PreConfiguredBackendPoolLoadBalancerTypesAll
 	bi = newBackendPoolTypeNodeIP(az)
-	preConfigured, changed, shouldRefresh, err := bi.ReconcileBackendPools(testClusterName, &service, lb)
+	preConfigured, changed, err := bi.ReconcileBackendPools(testClusterName, &service, lb)
 	assert.NoError(t, err)
 	assert.False(t, preConfigured)
-	assert.False(t, shouldRefresh)
 	assert.True(t, changed)
 }
 
@@ -568,15 +570,18 @@ func TestReconcileBackendPoolsNodeIPEmptyPool(t *testing.T) {
 	mockVMSet.EXPECT().EnsureBackendPoolDeleted(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil)
 	mockVMSet.EXPECT().GetPrimaryVMSetName().Return("k8s-agentpool1-00000000")
 
+	lbClient := mockloadbalancerclient.NewMockInterface(ctrl)
+	lbClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(network.LoadBalancer{}, nil)
+
 	az := GetTestCloud(ctrl)
 	az.LoadBalancerBackendPoolConfigurationType = consts.LoadBalancerBackendPoolConfigurationTypeNodeIP
 	az.VMSet = mockVMSet
+	az.LoadBalancerClient = lbClient
 	bi := newBackendPoolTypeNodeIP(az)
 
 	service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
 
-	_, _, shouldRefresh, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
-	assert.True(t, shouldRefresh)
+	_, _, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
 	assert.NoError(t, err)
 }
 
@@ -596,10 +601,9 @@ func TestReconcileBackendPoolsNodeIPPreConfigured(t *testing.T) {
 
 	service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
 	bi := newBackendPoolTypeNodeIP(az)
-	preConfigured, changed, shouldRefresh, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
+	preConfigured, changed, err := bi.ReconcileBackendPools("kubernetes", &service, lb)
 	assert.True(t, preConfigured)
 	assert.False(t, changed)
-	assert.False(t, shouldRefresh)
 	assert.NoError(t, err)
 }
 
@@ -619,7 +623,7 @@ func TestReconcileBackendPoolsNodeIPConfigToIP(t *testing.T) {
 	az.VMSet = mockVMSet
 	bi := newBackendPoolTypeNodeIP(az)
 	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
-	_, _, _, err := bi.ReconcileBackendPools(testClusterName, &svc, &lb)
+	_, _, err := bi.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.Contains(t, err.Error(), "delete LB backend pool error")
 
 	lb = buildDefaultTestLB(testClusterName, []string{
@@ -627,47 +631,11 @@ func TestReconcileBackendPoolsNodeIPConfigToIP(t *testing.T) {
 		"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool2-00000000-nic-1/ipConfigurations/ipconfig1",
 	})
 	mockVMSet.EXPECT().EnsureBackendPoolDeleted(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil)
-	_, _, shouldRefresh, err := bi.ReconcileBackendPools(testClusterName, &svc, &lb)
+	_, _, err = bi.ReconcileBackendPools(testClusterName, &svc, &lb)
 	assert.NoError(t, err)
-	assert.True(t, shouldRefresh)
 	assert.Empty(t, (*lb.BackendAddressPools)[0].LoadBalancerBackendAddresses)
 }
 
-func TestReconcileBackendPoolsNodeIPConfigToIPWithMigrationAPI(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	lb := buildDefaultTestLB(testClusterName, []string{
-		"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool1-00000000-nic-1/ipConfigurations/ipconfig1",
-		"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool2-00000000-nic-1/ipConfigurations/ipconfig1",
-	})
-
-	mockVMSet := NewMockVMSet(ctrl)
-	mockVMSet.EXPECT().EnsureBackendPoolDeleted(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil)
-	mockVMSet.EXPECT().GetPrimaryVMSetName().Return("k8s-agentpool1-00000000").AnyTimes()
-
-	mockLBClient := mockloadbalancerclient.NewMockInterface(ctrl)
-	mockLBClient.EXPECT().MigrateToIPBasedBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(retry.NewError(false, errors.New("error")))
-
-	az := GetTestCloud(ctrl)
-	az.VMSet = mockVMSet
-	az.LoadBalancerClient = mockLBClient
-	az.EnableMigrateToIPBasedBackendPoolAPI = true
-
-	bi := newBackendPoolTypeNodeIP(az)
-	svc := getTestService("test", v1.ProtocolTCP, nil, false, 80)
-	_, _, _, err := bi.ReconcileBackendPools(testClusterName, &svc, &lb)
-	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "error")
-
-	mockLBClient.EXPECT().MigrateToIPBasedBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
-	bps := buildLBWithVMIPs(testClusterName, []string{"1.2.3.4", "2.3.4.5"}).BackendAddressPools
-	mockLBClient.EXPECT().GetLBBackendPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return((*bps)[0], nil)
-	_, _, shouldRefresh, err := bi.ReconcileBackendPools(testClusterName, &svc, &lb)
-	assert.NoError(t, err)
-	assert.True(t, shouldRefresh)
-}
-
 func buildTestLoadBalancerBackendPoolWithIPs(name string, ips []string) network.BackendAddressPool {
 	backendPool := network.BackendAddressPool{
 		Name: &name,
diff --git a/pkg/provider/azure_loadbalancer_test.go b/pkg/provider/azure_loadbalancer_test.go
index 829b8f34e..3b4dbb670 100644
--- a/pkg/provider/azure_loadbalancer_test.go
+++ b/pkg/provider/azure_loadbalancer_test.go
@@ -861,7 +861,7 @@ func TestEnsureLoadBalancerDeleted(t *testing.T) {
 	defer ctrl.Finish()
 	az := GetTestCloud(ctrl)
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -1845,6 +1845,38 @@ func TestGetServiceLoadBalancer(t *testing.T) {
 			expectedExists: false,
 			expectedError:  false,
 		},
+		{
+			desc:    "getServiceLoadBalancer should create a new lb and return the status of the previous lb",
+			sku:     "Basic",
+			wantLB:  true,
+			service: getTestService("service1", v1.ProtocolTCP, map[string]string{consts.ServiceAnnotationLoadBalancerMode: "as", consts.ServiceAnnotationLoadBalancerInternal: consts.TrueAnnotationValue}, false, 80),
+			existingLBs: []network.LoadBalancer{
+				{
+					Name:     pointer.String("as-internal"),
+					Location: pointer.String("westus"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						FrontendIPConfigurations: &[]network.FrontendIPConfiguration{
+							{
+								Name: pointer.String("aservice1"),
+								ID:   pointer.String("as-internal-aservice1"),
+								FrontendIPConfigurationPropertiesFormat: &network.FrontendIPConfigurationPropertiesFormat{
+									PrivateIPAddress: pointer.String("1.2.3.4"),
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedLB: &network.LoadBalancer{
+				Name:     pointer.String("testCluster-internal"),
+				Location: pointer.String("westus"),
+				Sku: &network.LoadBalancerSku{
+					Name: network.LoadBalancerSkuNameBasic,
+				},
+				LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{},
+			},
+			expectedStatus: &v1.LoadBalancerStatus{Ingress: []v1.LoadBalancerIngress{{IP: "1.2.3.4", Hostname: ""}}},
+		},
 	}
 
 	ctrl := gomock.NewController(t)
@@ -3606,19 +3638,17 @@ func TestReconcileLoadBalancerCommon(t *testing.T) {
 	}
 
 	testCases := []struct {
-		desc                                      string
-		service                                   v1.Service
-		loadBalancerSku                           string
-		preConfigLBType                           string
-		loadBalancerResourceGroup                 string
-		disableOutboundSnat                       *bool
-		wantLb                                    bool
-		shouldRefreshLBAfterReconcileBackendPools bool
-		existingLB                                network.LoadBalancer
-		expectedLB                                network.LoadBalancer
-		expectLBUpdate                            bool
-		expectedGetLBError                        *retry.Error
-		expectedError                             error
+		desc                      string
+		service                   v1.Service
+		loadBalancerSku           string
+		preConfigLBType           string
+		loadBalancerResourceGroup string
+		disableOutboundSnat       *bool
+		wantLb                    bool
+		existingLB                network.LoadBalancer
+		expectedLB                network.LoadBalancer
+		expectLBUpdate            bool
+		expectedError             error
 	}{
 		{
 			desc: "reconcileLoadBalancer shall return the lb deeply equal to the existingLB if there's no " +
@@ -3714,25 +3744,6 @@ func TestReconcileLoadBalancerCommon(t *testing.T) {
 			expectLBUpdate:            true,
 			expectedError:             nil,
 		},
-		{
-			desc:            "reconcileLoadBalancer should refresh the LB after reconciling backend pools if needed",
-			loadBalancerSku: "basic",
-			service:         service1,
-			existingLB:      basicLb1,
-			wantLb:          true,
-			shouldRefreshLBAfterReconcileBackendPools: true,
-			expectedLB:    basicLb1,
-			expectedError: nil,
-		},
-		{
-			desc:       "reconcileLoadBalancer should return the error if failed to get the LB after reconcileBackendPools",
-			service:    service1,
-			existingLB: basicLb1,
-			wantLb:     true,
-			shouldRefreshLBAfterReconcileBackendPools: true,
-			expectedGetLBError:                        retry.NewError(false, errors.New("error")),
-			expectedError:                             fmt.Errorf("reconcileLoadBalancer for service (default/service1): failed to get load balancer testCluster: %w", retry.NewError(false, errors.New("error")).Error()),
-		},
 	}
 
 	for _, test := range testCases {
@@ -3771,7 +3782,7 @@ func TestReconcileLoadBalancerCommon(t *testing.T) {
 
 			mockLBsClient := mockloadbalancerclient.NewMockInterface(ctrl)
 			mockLBsClient.EXPECT().List(gomock.Any(), az.getLoadBalancerResourceGroup()).Return([]network.LoadBalancer{test.existingLB}, nil)
-			mockLBsClient.EXPECT().Get(gomock.Any(), az.getLoadBalancerResourceGroup(), *test.existingLB.Name, gomock.Any()).Return(test.existingLB, test.expectedGetLBError).AnyTimes()
+			mockLBsClient.EXPECT().Get(gomock.Any(), az.getLoadBalancerResourceGroup(), *test.existingLB.Name, gomock.Any()).Return(test.existingLB, nil).AnyTimes()
 			expectLBUpdateCount := 1
 			if test.expectLBUpdate {
 				expectLBUpdateCount++
@@ -3783,10 +3794,7 @@ func TestReconcileLoadBalancerCommon(t *testing.T) {
 			assert.NoError(t, err.Error())
 
 			mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-			if test.shouldRefreshLBAfterReconcileBackendPools {
-				mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, true, nil)
-			}
-			mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+			mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 			mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 			lb, rerr := az.reconcileLoadBalancer("testCluster", &service, clusterResources.nodes, test.wantLb)
@@ -6358,6 +6366,47 @@ func TestEnsureLoadBalancerTagged(t *testing.T) {
 	}
 }
 
+func TestShouldChangeLoadBalancer(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	cloud := GetTestCloud(ctrl)
+	cloud.LoadBalancerSku = consts.LoadBalancerSkuBasic
+
+	t.Run("shouldChangeLoadBalancer should return true if the mode is different from the current vm set", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationLoadBalancerMode: "as2",
+		}
+		service := getTestService("service1", v1.ProtocolTCP, annotations, false, 80)
+		res := cloud.shouldChangeLoadBalancer(&service, "as1", "testCluster")
+		assert.True(t, res)
+	})
+
+	t.Run("shouldChangeLoadBalancer should return false if the current lb is the primary slb and the vmSet selected by annotation is sharing the primary slb", func(t *testing.T) {
+		cloud.LoadBalancerSku = consts.LoadBalancerSkuStandard
+		cloud.EnableMultipleStandardLoadBalancers = true
+		cloud.NodePoolsWithoutDedicatedSLB = "vmss-1,vmss2"
+
+		annotations := map[string]string{
+			consts.ServiceAnnotationLoadBalancerMode: "vmss-1",
+		}
+		service := getTestService("service1", v1.ProtocolTCP, annotations, false, 80)
+		res := cloud.shouldChangeLoadBalancer(&service, "testCluster-internal", "testCluster")
+		assert.False(t, res)
+	})
+
+	t.Run("shouldChangeLoadBalancer should return true if the mode is the same as the current LB but the vmSet is the primary one", func(t *testing.T) {
+		cloud.LoadBalancerSku = consts.LoadBalancerSkuStandard
+		cloud.EnableMultipleStandardLoadBalancers = true
+		cloud.PrimaryAvailabilitySetName = "vmss-1"
+		annotations := map[string]string{
+			consts.ServiceAnnotationLoadBalancerMode: "vmss-1",
+		}
+		service := getTestService("service1", v1.ProtocolTCP, annotations, false, 80)
+		res := cloud.shouldChangeLoadBalancer(&service, "vmss-1", "testCluster")
+		assert.True(t, res)
+	})
+}
+
 func TestRemoveFrontendIPConfigurationFromLoadBalancerDelete(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -6681,6 +6730,223 @@ func TestReconcileZonesForFrontendIPConfigs(t *testing.T) {
 	}
 }
 
+func TestReconcileSharedLoadBalancer(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	for _, tc := range []struct {
+		description, vmSetsSharingPrimarySLB                                                                  string
+		nodes                                                                                                 []*v1.Node
+		useMultipleSLBs, useBasicLB, useVMIP                                                                  bool
+		existingLBs                                                                                           []network.LoadBalancer
+		listLBErr                                                                                             *retry.Error
+		expectedListCount, expectedDeleteCount, expectedGetNamesCount, expectedCreateOrUpdateBackendPoolCount int
+		expectedLBs                                                                                           []network.LoadBalancer
+		expectedErr                                                                                           error
+	}{
+		{
+			description:             "reconcileSharedLoadBalancer should decouple the vmSet from its dedicated lb if the vmSet is sharing the primary slb",
+			useMultipleSLBs:         true,
+			vmSetsSharingPrimarySLB: "vmss1,vmss2",
+			nodes: []*v1.Node{
+				{ObjectMeta: metav1.ObjectMeta{Name: "kubernetes"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "vmss1"}},
+			},
+			existingLBs: []network.LoadBalancer{
+				{
+					Name: pointer.String("kubernetes"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss2-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Name: pointer.String("kubernetes-internal"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss2-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Name: pointer.String("vmss1"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss1-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Name: pointer.String("vmss1-internal"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss1-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedLBs: []network.LoadBalancer{
+				{
+					Name: pointer.String("kubernetes"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss2-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					Name: pointer.String("kubernetes-internal"),
+					LoadBalancerPropertiesFormat: &network.LoadBalancerPropertiesFormat{
+						BackendAddressPools: &[]network.BackendAddressPool{
+							{
+								Name: pointer.String("kubernetes"),
+								BackendAddressPoolPropertiesFormat: &network.BackendAddressPoolPropertiesFormat{
+									BackendIPConfigurations: &[]network.InterfaceIPConfiguration{
+										{
+											ID: pointer.String("vmss2-nic-1"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedListCount:     1,
+			expectedGetNamesCount: 1,
+			expectedDeleteCount:   1,
+		},
+		{
+			description:       "reconcileSharedLoadBalancer should do nothing if the basic load balancer is used",
+			useBasicLB:        true,
+			expectedListCount: 1,
+		},
+		{
+			description:       "reconcileSharedLoadBalancer should do nothing if `nodes` is nil",
+			expectedListCount: 1,
+		},
+		{
+			description:     "reconcileSharedLoadBalancer should do nothing if the vmSet is not sharing the primary slb",
+			useMultipleSLBs: true,
+			nodes: []*v1.Node{
+				{ObjectMeta: metav1.ObjectMeta{Name: "kubernetes"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "vmss1"}},
+			},
+			existingLBs: []network.LoadBalancer{
+				{
+					Name: pointer.String("kubernetes"),
+				},
+				{
+					Name: pointer.String("vmss1"),
+				},
+			},
+			expectedLBs: []network.LoadBalancer{
+				{
+					Name: pointer.String("kubernetes"),
+				},
+				{
+					Name: pointer.String("vmss1"),
+				},
+			},
+			expectedListCount:     1,
+			expectedGetNamesCount: 1,
+		},
+		{
+			description: "reconcileSharedLoadBalancer should report an error if failed to list managed load balancers",
+			nodes: []*v1.Node{
+				{ObjectMeta: metav1.ObjectMeta{Name: "kubernetes"}},
+				{ObjectMeta: metav1.ObjectMeta{Name: "vmss1"}},
+			},
+			expectedListCount: 1,
+			listLBErr:         retry.NewError(false, errors.New("error")),
+			expectedErr:       errors.New("reconcileSharedLoadBalancer: failed to list managed LB: Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: error"),
+		},
+	} {
+		t.Run(tc.description, func(t *testing.T) {
+			cloud := GetTestCloud(ctrl)
+
+			cloud.NodePoolsWithoutDedicatedSLB = tc.vmSetsSharingPrimarySLB
+
+			cloud.LoadBalancerSku = consts.VMTypeStandard
+			if tc.useMultipleSLBs {
+				cloud.EnableMultipleStandardLoadBalancers = true
+			} else if tc.useBasicLB {
+				cloud.LoadBalancerSku = consts.LoadBalancerSkuBasic
+			}
+
+			mockLBClient := cloud.LoadBalancerClient.(*mockloadbalancerclient.MockInterface)
+			mockLBClient.EXPECT().List(gomock.Any(), cloud.ResourceGroup).Return(tc.existingLBs, tc.listLBErr).Times(tc.expectedListCount)
+			mockLBClient.EXPECT().Delete(gomock.Any(), cloud.ResourceGroup, "vmss1").Return(nil).Times(tc.expectedDeleteCount)
+			mockLBClient.EXPECT().Delete(gomock.Any(), cloud.ResourceGroup, "vmss1-internal").Return(nil).Times(tc.expectedDeleteCount)
+
+			if tc.useVMIP {
+				cloud.LoadBalancerBackendPoolConfigurationType = consts.LoadBalancerBackendPoolConfigurationTypeNodeIP
+				tc.expectedDeleteCount = 0
+			}
+
+			mockVMSet := NewMockVMSet(ctrl)
+			mockVMSet.EXPECT().EnsureBackendPoolDeleted(gomock.Any(), []string{"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/vmss1/backendAddressPools/kubernetes"}, "vmss1", gomock.Any(), gomock.Any()).Return(false, nil).Times(tc.expectedDeleteCount)
+			mockVMSet.EXPECT().EnsureBackendPoolDeleted(gomock.Any(), []string{"/subscriptions/subscription/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/vmss1-internal/backendAddressPools/kubernetes"}, "vmss1", gomock.Any(), gomock.Any()).Return(false, nil).Times(tc.expectedDeleteCount)
+			mockVMSet.EXPECT().GetAgentPoolVMSetNames(gomock.Any()).Return(&[]string{"vmss1", "vmss2"}, nil).MaxTimes(tc.expectedGetNamesCount)
+			mockVMSet.EXPECT().GetPrimaryVMSetName().Return("vmss2").AnyTimes()
+			cloud.VMSet = mockVMSet
+
+			service := getTestService("test", v1.ProtocolTCP, nil, false, 80)
+			lbs, err := cloud.reconcileSharedLoadBalancer(&service, "kubernetes", tc.nodes)
+			if tc.expectedErr != nil {
+				assert.Equal(t, tc.expectedErr.Error(), err.Error())
+			}
+			assert.Equal(t, tc.expectedLBs, lbs)
+		})
+	}
+}
+
 func TestReconcileIPSettings(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
diff --git a/pkg/provider/azure_standard_test.go b/pkg/provider/azure_standard_test.go
index 9e057a3ba..3944f3beb 100644
--- a/pkg/provider/azure_standard_test.go
+++ b/pkg/provider/azure_standard_test.go
@@ -235,13 +235,15 @@ func TestGetAzureLoadBalancerName(t *testing.T) {
 	az.PrimaryAvailabilitySetName = primary
 
 	cases := []struct {
-		description   string
-		vmSet         string
-		isInternal    bool
-		useStandardLB bool
-		clusterName   string
-		lbName        string
-		expected      string
+		description             string
+		vmSet                   string
+		isInternal              bool
+		useStandardLB           bool
+		enableMultipleSLBs      bool
+		vmSetsSharingPrimarySLB string
+		clusterName             string
+		lbName                  string
+		expected                string
 	}{
 		{
 			description: "prefix of loadBalancerName should be az.LoadBalancerName if az.LoadBalancerName is not nil",
@@ -306,6 +308,23 @@ func TestGetAzureLoadBalancerName(t *testing.T) {
 			clusterName:   "azure",
 			expected:      "azure-internal",
 		},
+		{
+			description:        "getAzureLoadBalancerName should return the vmSet name if multiple slbs are enabled",
+			vmSet:              "as",
+			useStandardLB:      true,
+			enableMultipleSLBs: true,
+			clusterName:        "azure",
+			expected:           "as",
+		},
+		{
+			description:             "getAzureLoadBalancerName should return the cluster name if multiple slbs are enabled and the vmSet is sharing the primary slb",
+			vmSet:                   "as",
+			useStandardLB:           true,
+			enableMultipleSLBs:      true,
+			vmSetsSharingPrimarySLB: "as , as-1",
+			clusterName:             "azure",
+			expected:                "azure",
+		},
 	}
 
 	for _, c := range cases {
@@ -314,7 +333,12 @@ func TestGetAzureLoadBalancerName(t *testing.T) {
 		} else {
 			az.Config.LoadBalancerSku = consts.LoadBalancerSkuBasic
 		}
-
+		if c.enableMultipleSLBs {
+			az.EnableMultipleStandardLoadBalancers = true
+		}
+		if c.vmSetsSharingPrimarySLB != "" {
+			az.NodePoolsWithoutDedicatedSLB = c.vmSetsSharingPrimarySLB
+		}
 		az.Config.LoadBalancerName = c.lbName
 		loadbalancerName := az.getAzureLoadBalancerName(c.clusterName, c.vmSet, c.isInternal)
 		assert.Equal(t, c.expected, loadbalancerName, c.description)
@@ -1556,6 +1580,7 @@ func TestGetStandardVMSetNames(t *testing.T) {
 	for _, test := range testCases {
 		cloud := GetTestCloud(ctrl)
 		if test.usingSingleSLBS {
+			cloud.EnableMultipleStandardLoadBalancers = false
 			cloud.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 		mockVMClient := cloud.VirtualMachinesClient.(*mockvmclient.MockInterface)
@@ -1611,6 +1636,7 @@ func TestStandardEnsureHostInPool(t *testing.T) {
 		vmSetName         string
 		nicProvisionState network.ProvisioningState
 		isStandardLB      bool
+		useMultipleSLBs   bool
 		expectedErrMsg    error
 	}{
 		{
@@ -1621,6 +1647,16 @@ func TestStandardEnsureHostInPool(t *testing.T) {
 			nicID:     "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/nic1",
 			vmSetName: "availabilityset-1",
 		},
+		{
+			name:            "EnsureHostInPool should return nil if node is not in VMSet when using multiple SLBs",
+			service:         &v1.Service{},
+			nodeName:        "vm1",
+			nicName:         "nic1",
+			nicID:           "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/nic1",
+			vmSetName:       "availabilityset-1",
+			isStandardLB:    true,
+			useMultipleSLBs: true,
+		},
 		{
 			name:           "EnsureHostInPool should report error if last segment of nicID is nil",
 			service:        &v1.Service{},
@@ -1685,6 +1721,10 @@ func TestStandardEnsureHostInPool(t *testing.T) {
 			cloud.Config.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if test.useMultipleSLBs {
+			cloud.EnableMultipleStandardLoadBalancers = true
+		}
+
 		testVM := buildDefaultTestVirtualMachine(availabilitySetID, []string{test.nicID})
 		testVM.Name = pointer.String(string(test.nodeName))
 		testNIC := buildDefaultTestInterface(false, []string{backendAddressPoolID})
diff --git a/pkg/provider/azure_test.go b/pkg/provider/azure_test.go
index 0b6401127..6f02949e8 100644
--- a/pkg/provider/azure_test.go
+++ b/pkg/provider/azure_test.go
@@ -112,7 +112,7 @@ func TestAddPort(t *testing.T) {
 	setMockLBs(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -487,7 +487,7 @@ func testLoadBalancerServiceDefaultModeSelection(t *testing.T, isInternal bool)
 	setMockEnv(az, ctrl, expectedInterfaces, expectedVirtualMachines, serviceCount)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -556,7 +556,7 @@ func testLoadBalancerServiceAutoModeSelection(t *testing.T, isInternal bool) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -637,7 +637,7 @@ func testLoadBalancerServicesSpecifiedSelection(t *testing.T, isInternal bool) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -696,7 +696,7 @@ func testLoadBalancerMaxRulesServices(t *testing.T, isInternal bool) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -781,7 +781,7 @@ func testLoadBalancerServiceAutoModeDeleteSelection(t *testing.T, isInternal boo
 	expectedLBs := make([]network.LoadBalancer, 0)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockLBBackendPool.EXPECT().GetBackendPrivateIPs(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
 
@@ -875,7 +875,7 @@ func TestReconcileLoadBalancerAddServiceOnInternalSubnet(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, true)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -939,7 +939,7 @@ func TestReconcileLoadBalancerAddServicesOnMultipleSubnets(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	// svc1 is using LB without "-internal" suffix
@@ -987,7 +987,7 @@ func TestReconcileLoadBalancerEditServiceSubnet(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, true)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	expectedPLS := make([]network.PrivateLinkService, 0)
@@ -1034,7 +1034,7 @@ func TestReconcileLoadBalancerNodeHealth(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1061,7 +1061,7 @@ func TestReconcileLoadBalancerRemoveService(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	_, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1099,7 +1099,7 @@ func TestReconcileLoadBalancerRemoveAllPortsRemovesFrontendConfig(t *testing.T)
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1135,7 +1135,7 @@ func TestReconcileLoadBalancerRemovesPort(t *testing.T) {
 	setMockEnvDualStack(az, ctrl, expectedInterfaces, expectedVirtualMachines, 1)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	expectedLBs := make([]network.LoadBalancer, 0)
@@ -1169,7 +1169,7 @@ func TestReconcileLoadBalancerMultipleServices(t *testing.T) {
 	setMockLBsDualStack(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	expectedPLS := make([]network.PrivateLinkService, 0)
@@ -1231,7 +1231,7 @@ func TestServiceDefaultsToNoSessionPersistence(t *testing.T) {
 	mockPIPsClient.EXPECT().List(gomock.Any(), az.ResourceGroup).Return([]network.PublicIPAddress{expectedPIP}, nil).AnyTimes()
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1290,7 +1290,7 @@ func TestServiceRespectsNoSessionAffinity(t *testing.T) {
 	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1351,7 +1351,7 @@ func TestServiceRespectsClientIPSessionAffinity(t *testing.T) {
 	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
@@ -1383,7 +1383,7 @@ func TestReconcileSecurityGroupNewServiceAddsPort(t *testing.T) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 	setMockLBs(az, ctrl, &expectedLBs, "service", 1, 1, false)
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	lb, _ := az.reconcileLoadBalancer(testClusterName, &svc1, clusterResources.nodes, true)
 	lbStatus, _, _ := az.getServiceLoadBalancerStatus(&svc1, lb)
@@ -1409,7 +1409,7 @@ func TestReconcileSecurityGroupNewInternalServiceAddsPort(t *testing.T) {
 	setMockLBs(az, ctrl, &expectedLBs, "service", 1, 1, true)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, _ := az.reconcileLoadBalancer(testClusterName, &svc1, clusterResources.nodes, true)
@@ -1442,7 +1442,7 @@ func TestReconcileSecurityGroupRemoveService(t *testing.T) {
 			"Standard"), nil).AnyTimes()
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, _ := az.reconcileLoadBalancer(testClusterName, &service1, clusterResources.nodes, true)
@@ -1469,7 +1469,7 @@ func TestReconcileSecurityGroupRemoveServiceRemovesPort(t *testing.T) {
 	setMockEnv(az, ctrl, expectedInterfaces, expectedVirtualMachines, 1)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	getTestSecurityGroup(az, svc)
@@ -1507,7 +1507,7 @@ func TestReconcileSecurityWithSourceRanges(t *testing.T) {
 	setMockEnv(az, ctrl, expectedInterfaces, expectedVirtualMachines, 1)
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	getTestSecurityGroup(az, svc)
@@ -1556,7 +1556,7 @@ func TestReconcileSecurityGroupEtagMismatch(t *testing.T) {
 			"Standard"), nil).AnyTimes()
 
 	mockLBBackendPool := az.LoadBalancerBackendPool.(*MockBackendPool)
-	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, false, nil).AnyTimes()
+	mockLBBackendPool.EXPECT().ReconcileBackendPools(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, false, nil).AnyTimes()
 	mockLBBackendPool.EXPECT().EnsureHostsInPool(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 
 	lb, _ := az.reconcileLoadBalancer(testClusterName, &svc1, clusterResources.nodes, true)
diff --git a/pkg/provider/azure_vmss_test.go b/pkg/provider/azure_vmss_test.go
index 4f9b34fa5..26e6d612f 100644
--- a/pkg/provider/azure_vmss_test.go
+++ b/pkg/provider/azure_vmss_test.go
@@ -1663,6 +1663,7 @@ func TestGetVMSetNames(t *testing.T) {
 		assert.NoError(t, err, "unexpected error when creating test VMSS")
 
 		if test.useSingleSLB {
+			ss.EnableMultipleStandardLoadBalancers = false
 			ss.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
@@ -2169,6 +2170,7 @@ func TestEnsureHostInPool(t *testing.T) {
 		vmSetName                 string
 		isBasicLB                 bool
 		isNilVMNetworkConfigs     bool
+		useMultipleSLBs           bool
 		isVMBeingDeleted          bool
 		expectedNodeResourceGroup string
 		expectedVMSSName          string
@@ -2183,6 +2185,12 @@ func TestEnsureHostInPool(t *testing.T) {
 			vmSetName:   "vmss-1",
 			isBasicLB:   true,
 		},
+		{
+			description:     "EnsureHostInPool should skip the current node if the vmSetName is not equal to the node's vmss name and multiple SLBs are used",
+			nodeName:        "vmss-vm-000000",
+			vmSetName:       "vmss-1",
+			useMultipleSLBs: true,
+		},
 		{
 			description:           "EnsureHostInPool should skip the current node if the network configs of the VMSS VM is nil",
 			nodeName:              "vmss-vm-000000",
@@ -2263,6 +2271,10 @@ func TestEnsureHostInPool(t *testing.T) {
 			ss.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if test.useMultipleSLBs {
+			ss.EnableMultipleStandardLoadBalancers = true
+		}
+
 		expectedVMSS := buildTestVMSS(testVMSSName, "vmss-vm-")
 		mockVMSSClient := ss.cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 		mockVMSSClient.EXPECT().List(gomock.Any(), ss.ResourceGroup).Return([]compute.VirtualMachineScaleSet{expectedVMSS}, nil).AnyTimes()
diff --git a/pkg/provider/azure_vmssflex_test.go b/pkg/provider/azure_vmssflex_test.go
index 40bd53549..2d70489f5 100644
--- a/pkg/provider/azure_vmssflex_test.go
+++ b/pkg/provider/azure_vmssflex_test.go
@@ -224,6 +224,7 @@ func TestGetVMSetNamesVmssFlex(t *testing.T) {
 		fs.vmssFlexVMNameToVmssID.Store(testNodeName2, testVmssFlexID2)
 
 		if tc.useSingleSLB {
+			fs.EnableMultipleStandardLoadBalancers = false
 			fs.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
@@ -941,6 +942,8 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 		vmSetNameOfLB                  string
 		backendPoolID                  string
 		isStandardLB                   bool
+		useMultipleSLBs                bool
+		NodePoolsWithoutDedicatedSLB   string
 		testVMListWithoutInstanceView  []compute.VirtualMachine
 		testVMListWithOnlyInstanceView []compute.VirtualMachine
 		vmListErr                      error
@@ -959,6 +962,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -976,6 +980,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -986,6 +991,61 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			expectedNodeName:               "",
 			expectedErr:                    nil,
 		},
+		{
+			description:                    "EnsureHostInPool should return error if basic load balancer is used",
+			nodeName:                       "vmssflex1000001",
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   false,
+			useMultipleSLBs:                true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			nic:                            network.Interface{},
+			nicGetErr:                      nil,
+			expectedNodeResourceGroup:      "",
+			expectedVMSetName:              "",
+			expectedNodeName:               "",
+			expectedErr:                    fmt.Errorf("EnsureHostInPool: VMSS Flex does not support Basic Load Balancer"),
+		},
+		{
+			description:                    "EnsureHostInPool should skip the current node if the vmSetName is not equal to the node's vmss name and multiple SLBs are used",
+			nodeName:                       "vmssflex1000001",
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "anotherVmssFlex",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			nic:                            network.Interface{},
+			nicGetErr:                      nil,
+			expectedNodeResourceGroup:      "",
+			expectedVMSetName:              "",
+			expectedNodeName:               "",
+			expectedErr:                    errNotInVMSet,
+		},
+		{
+			description:                    "EnsureHostInPool should add a new backend pool to the vm if the vmss name is in the sharingPrimaryVMSetName",
+			nodeName:                       "vmssflex1000001",
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "vmss",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
+			NodePoolsWithoutDedicatedSLB:   "vmssflex1",
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			nic:                            generateTestNic("testvm1-nic", false, network.ProvisioningStateSucceeded, "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/testvm1"),
+			nicGetErr:                      nil,
+			expectedNodeResourceGroup:      "rg",
+			expectedVMSetName:              "vmssflex1",
+			expectedNodeName:               "vmssflex1000001",
+			expectedErr:                    nil,
+		},
 		{
 			description:                    "EnsureHostInPool should skip the current node if the network configs of the VMSS VM is nil",
 			nodeName:                       "vmssflex1000001",
@@ -993,6 +1053,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1010,6 +1071,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1027,6 +1089,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1045,6 +1108,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1063,6 +1127,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  testBackendPoolID0,
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1080,6 +1145,7 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb2-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1099,6 +1165,11 @@ func TestEnsureHostInPoolVmssFlex(t *testing.T) {
 			fs.Config.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if tc.useMultipleSLBs {
+			fs.EnableMultipleStandardLoadBalancers = true
+		}
+		fs.NodePoolsWithoutDedicatedSLB = tc.NodePoolsWithoutDedicatedSLB
+
 		mockVMSSClient := fs.cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 		mockVMSSClient.EXPECT().List(gomock.Any(), gomock.Any()).Return(testVmssFlexList, nil).AnyTimes()
 
@@ -1133,6 +1204,7 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 		vmSetNameOfLB                  string
 		backendPoolID                  string
 		isStandardLB                   bool
+		useMultipleSLBs                bool
 		isVMSSDeallocating             bool
 		hasDefaultVMProfile            bool
 		testVMListWithoutInstanceView  []compute.VirtualMachine
@@ -1154,6 +1226,7 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			hasDefaultVMProfile:            true,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
@@ -1173,6 +1246,28 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
+			hasDefaultVMProfile:            true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			expectedErr:                    nil,
+		},
+		{
+			description: "ensureVMSSFlexInPool should skip the node if the corresponding VMSS is deallocating",
+			nodes: []*v1.Node{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vmssflex1000001",
+					},
+				},
+			},
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   true,
+			useMultipleSLBs:                false,
+			isVMSSDeallocating:             true,
 			hasDefaultVMProfile:            true,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
@@ -1192,6 +1287,7 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			isVMSSDeallocating:             false,
 			hasDefaultVMProfile:            false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
@@ -1212,6 +1308,7 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  testBackendPoolID0,
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			hasDefaultVMProfile:            true,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
@@ -1231,12 +1328,74 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb2-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			hasDefaultVMProfile:            true,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
 			expectedErr:                    nil,
 		},
+		{
+			description: "ensureVMSSFlexInPool should not work for the basic load balancer",
+			nodes: []*v1.Node{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vmssflex1000001",
+					},
+				},
+			},
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   false,
+			useMultipleSLBs:                true,
+			hasDefaultVMProfile:            true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			expectedErr:                    fmt.Errorf("ensureVMSSFlexInPool: VMSS Flex does not support Basic Load Balancer"),
+		},
+		{
+			description: "ensureVMSSFlexInPool should work for multiple standard load balancers",
+			nodes: []*v1.Node{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vmssflex1000001",
+					},
+				},
+			},
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "vmssflex1",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
+			hasDefaultVMProfile:            true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			expectedErr:                    nil,
+		},
+		{
+			description: "ensureVMSSFlexInPool return error if failing to update nic",
+			nodes: []*v1.Node{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "vmssflex1000001",
+					},
+				},
+			},
+			service:                        &v1.Service{},
+			vmSetNameOfLB:                  "vmssflex1",
+			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
+			hasDefaultVMProfile:            true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			vmssPutErr:                     &retry.Error{RawError: fmt.Errorf("failed to update nic")},
+			expectedErr:                    fmt.Errorf("Retriable: false, RetryAfter: 0s, HTTPStatusCode: 0, RawError: failed to update nic"),
+		},
 	}
 
 	for _, tc := range testCases {
@@ -1246,6 +1405,10 @@ func TestEnsureVMSSFlexInPool(t *testing.T) {
 			fs.Config.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if tc.useMultipleSLBs {
+			fs.EnableMultipleStandardLoadBalancers = true
+		}
+
 		testVmssFlex := genreteTestVmssFlex("vmssflex1", testVmssFlex1ID)
 
 		if tc.isVMSSDeallocating {
@@ -1285,6 +1448,7 @@ func TestEnsureHostsInPoolVmssFlex(t *testing.T) {
 		vmSetNameOfLB                  string
 		backendPoolID                  string
 		isStandardLB                   bool
+		useMultipleSLBs                bool
 		testVMListWithoutInstanceView  []compute.VirtualMachine
 		testVMListWithOnlyInstanceView []compute.VirtualMachine
 		vmListErr                      error
@@ -1306,6 +1470,7 @@ func TestEnsureHostsInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1326,6 +1491,7 @@ func TestEnsureHostsInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   false,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1346,6 +1512,7 @@ func TestEnsureHostsInPoolVmssFlex(t *testing.T) {
 			vmSetNameOfLB:                  "",
 			backendPoolID:                  "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/loadBalancers/lb-internal/backendAddressPools/backendpool-1",
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1363,6 +1530,10 @@ func TestEnsureHostsInPoolVmssFlex(t *testing.T) {
 			fs.Config.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if tc.useMultipleSLBs {
+			fs.EnableMultipleStandardLoadBalancers = true
+		}
+
 		mockVMSSClient := fs.cloud.VirtualMachineScaleSetsClient.(*mockvmssclient.MockInterface)
 		mockVMSSClient.EXPECT().List(gomock.Any(), gomock.Any()).Return([]compute.VirtualMachineScaleSet{genreteTestVmssFlex("vmssflex1", testVmssFlex1ID)}, nil).AnyTimes()
 		mockVMSSClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(testVmssFlex1, nil).AnyTimes()
@@ -1624,6 +1795,7 @@ func TestEnsureBackendPoolDeletedVmssFlex(t *testing.T) {
 		backendAddressPools *[]network.BackendAddressPool
 		deleteFromVMSet     bool
 		isStandardLB        bool
+		useMultipleSLBs     bool
 
 		testVMListWithoutInstanceView  []compute.VirtualMachine
 		testVMListWithOnlyInstanceView []compute.VirtualMachine
@@ -1643,6 +1815,7 @@ func TestEnsureBackendPoolDeletedVmssFlex(t *testing.T) {
 			backendAddressPools:            testBackendPools,
 			deleteFromVMSet:                true,
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1658,6 +1831,39 @@ func TestEnsureBackendPoolDeletedVmssFlex(t *testing.T) {
 			backendAddressPools:            nil,
 			deleteFromVMSet:                true,
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			nic:                            generateTestNic("testvm1-nic", false, network.ProvisioningStateSucceeded, "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/testvm1"),
+			nicGetErr:                      nil,
+			expectedErr:                    nil,
+		},
+		{
+			description:                    "EnsureBackendPoolDeleted should delete a backend pool if the VMSetName matches nic's vmss",
+			service:                        &v1.Service{},
+			vmSetName:                      "vmssflex1",
+			backendPoolID:                  testBackendPoolID0,
+			backendAddressPools:            testBackendPools,
+			deleteFromVMSet:                true,
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
+			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
+			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
+			vmListErr:                      nil,
+			nic:                            generateTestNic("testvm1-nic", false, network.ProvisioningStateSucceeded, "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/testvm1"),
+			nicGetErr:                      nil,
+			expectedErr:                    nil,
+		},
+		{
+			description:                    "EnsureBackendPoolDeleted should do nothing if the VMSetName does not match nic's vmss",
+			service:                        &v1.Service{},
+			vmSetName:                      "vmssflex2",
+			backendPoolID:                  testBackendPoolID0,
+			backendAddressPools:            testBackendPools,
+			deleteFromVMSet:                true,
+			isStandardLB:                   true,
+			useMultipleSLBs:                true,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1673,6 +1879,7 @@ func TestEnsureBackendPoolDeletedVmssFlex(t *testing.T) {
 			backendAddressPools:            testBackendPools,
 			deleteFromVMSet:                true,
 			isStandardLB:                   true,
+			useMultipleSLBs:                false,
 			testVMListWithoutInstanceView:  testVMListWithoutInstanceView,
 			testVMListWithOnlyInstanceView: testVMListWithOnlyInstanceView,
 			vmListErr:                      nil,
@@ -1690,6 +1897,10 @@ func TestEnsureBackendPoolDeletedVmssFlex(t *testing.T) {
 			fs.Config.LoadBalancerSku = consts.LoadBalancerSkuStandard
 		}
 
+		if tc.useMultipleSLBs {
+			fs.EnableMultipleStandardLoadBalancers = true
+		}
+
 		testVmssFlex := genreteTestVmssFlex("vmssflex1", testVmssFlex1ID)
 		vmssFlexList := []compute.VirtualMachineScaleSet{testVmssFlex, genreteTestVmssFlex("vmssflex2", testVmssFlex2ID)}
 
diff --git a/tests/k8s-azure/manifest/cluster-api/cloud-config-vmss-default.json b/tests/k8s-azure/manifest/cluster-api/cloud-config-vmss-default.json
index d92278783..a90c956c3 100644
--- a/tests/k8s-azure/manifest/cluster-api/cloud-config-vmss-default.json
+++ b/tests/k8s-azure/manifest/cluster-api/cloud-config-vmss-default.json
@@ -18,6 +18,7 @@
   "useManagedIdentityExtension": false,
   "useInstanceMetadata": true,
   "primaryScaleSetName": "${CLUSTER_NAME}-vmss-0",
+  "enableMultipleStandardLoadBalancers": ${ENABLE_MULTI_SLB},
   "loadBalancerBackendPoolConfigurationType": "${LB_BACKEND_POOL_CONFIG_TYPE}",
   "putVMSSVMBatchSize": ${PUT_VMSS_VM_BATCH_SIZE},
   "enableVmssFlexNodes": ${ENABLE_VMSS_FLEX}