Repository: kubernetes-sigs/cloud-provider-azure
Pull request #1683 (2022-05-06 08:02:27)

[Release-1.0]feat: Private Link Service Integration for k8s service
<!--  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:
Cherry pick private link service integration feature to release-1.0 branch
#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #
related #1484 #1491 
#### 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
User can deploy managed Azure private link service resources by annotating the service object.
Detailed design doc can be found here: https://kubernetes-sigs.github.io/cloud-provider-azure/development/design-docs/pls-integration/
```

#### 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: Hi @jwtty. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fcloud-provider-azure).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/cloud-provider-azure/pull/1683#" title="Author self-approved">jwtty</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **martinforreal** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @martinforreal` in a comment when ready.

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).

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

- **[OWNERS](https://github.com/kubernetes-sigs/cloud-provider-azure/blob/release-1.0/OWNERS)**

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":["martinforreal"]} -->
MartinForReal: /ok-to-test
k8s-ci-robot: @jwtty: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-unit-1-21 | 6821ba8dd7cb73d0338d7127cd9e259453fbb685 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/1683/pull-cloud-provider-azure-unit-1-21/1522487287504441344) | true | `/test pull-cloud-provider-azure-unit-1-21`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=1683). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Ajwtty). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @jwtty: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-azure-unit-1-21 | 6821ba8dd7cb73d0338d7127cd9e259453fbb685 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/1683/pull-cloud-provider-azure-unit-1-21/1522487287504441344) | true | `/test pull-cloud-provider-azure-unit-1-21`
pull-cloud-provider-azure-e2e-ccm-vmss-1-21 | 6821ba8dd7cb73d0338d7127cd9e259453fbb685 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_cloud-provider-azure/1683/pull-cloud-provider-azure-e2e-ccm-vmss-1-21/1522487287391195136) | true | `/test pull-cloud-provider-azure-e2e-ccm-vmss-1-21`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=cloud-provider-azure&pr=1683). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Ajwtty). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/cloud-provider-azure/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/Dockerfile b/Dockerfile
index 275d795e2..51e4c9cee 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.18.1-buster AS builder
+FROM --platform=linux/amd64 golang:1.18-buster AS builder
 
 ARG ENABLE_GIT_COMMAND=true
 ARG ARCH=amd64
diff --git a/cloud-node-manager.Dockerfile b/cloud-node-manager.Dockerfile
index 09e5fdbbe..adaabdec7 100644
--- a/cloud-node-manager.Dockerfile
+++ b/cloud-node-manager.Dockerfile
@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.18.1-stretch AS builder
+FROM --platform=linux/amd64 golang:1.16.6-stretch AS builder
 
 ARG ENABLE_GIT_COMMAND=true
 ARG ARCH=amd64
diff --git a/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient.go b/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient.go
index 3a453cf89..5b6bee721 100644
--- a/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient.go
+++ b/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient.go
@@ -41,7 +41,8 @@ import (
 var _ Interface = &Client{}
 
 const (
-	PLSResourceType = "Microsoft.Network/privatelinkservices"
+	PLSResourceType    = "Microsoft.Network/privatelinkservices"
+	PEConnResourceType = "privateEndpointConnections"
 )
 
 // Client implements privatelinkservice Interface.
@@ -331,6 +332,50 @@ func (c *Client) deletePLS(ctx context.Context, resourceGroupName string, privat
 	return c.armClient.DeleteResource(ctx, resourceID, "")
 }
 
+func (c *Client) DeletePEConnection(ctx context.Context, resourceGroupName string, privateLinkServiceName string, privateEndpointConnectionName string) *retry.Error {
+	mc := metrics.NewMetricContext("private_endpoint_connection", "delete", resourceGroupName, c.subscriptionID, "")
+
+	// Report errors if the client is rate limited.
+	if !c.rateLimiterWriter.TryAccept() {
+		mc.RateLimitedCount()
+		return retry.GetRateLimitError(true, "PEConnDelete")
+	}
+
+	// Report errors if the client is throttled.
+	if c.RetryAfterWriter.After(time.Now()) {
+		mc.ThrottledCount()
+		rerr := retry.GetThrottlingError("PEConnDelete", "client throttled", c.RetryAfterWriter)
+		return rerr
+	}
+
+	rerr := c.deletePEConn(ctx, resourceGroupName, privateLinkServiceName, privateEndpointConnectionName)
+	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
+}
+
+// deletePLS deletes a private endpoint connection by name.
+func (c *Client) deletePEConn(ctx context.Context, resourceGroupName string, privateLinkServiceName string, privateEndpointConnectionName string) *retry.Error {
+	resourceID := armclient.GetChildResourceID(
+		c.subscriptionID,
+		resourceGroupName,
+		PLSResourceType,
+		privateLinkServiceName,
+		PEConnResourceType,
+		privateEndpointConnectionName,
+	)
+
+	return c.armClient.DeleteResource(ctx, resourceID, "")
+}
+
 func (c *Client) listResponder(resp *http.Response) (result network.PrivateLinkServiceListResult, err error) {
 	err = autorest.Respond(
 		resp,
diff --git a/pkg/azureclients/privatelinkserviceclient/interface.go b/pkg/azureclients/privatelinkserviceclient/interface.go
index 4a45c8618..b2cfefebb 100644
--- a/pkg/azureclients/privatelinkserviceclient/interface.go
+++ b/pkg/azureclients/privatelinkserviceclient/interface.go
@@ -48,4 +48,7 @@ type Interface interface {
 
 	// Delete deletes a private link service by name.
 	Delete(ctx context.Context, resourceGroupName string, privateLinkServiceName string) *retry.Error
+
+	// Delete deletes a private endpoint connection to the private link service by name
+	DeletePEConnection(ctx context.Context, resourceGroupName string, privateLinkServiceName string, privateEndpointConnectionName string) *retry.Error
 }
diff --git a/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient/interface.go b/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient/interface.go
index bb2dc90e8..1397982cd 100644
--- a/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient/interface.go
+++ b/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient/interface.go
@@ -82,6 +82,20 @@ func (mr *MockInterfaceMockRecorder) Delete(ctx, resourceGroupName, privateLinkS
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockInterface)(nil).Delete), ctx, resourceGroupName, privateLinkServiceName)
 }
 
+// DeletePEConnection mocks base method.
+func (m *MockInterface) DeletePEConnection(ctx context.Context, resourceGroupName, privateLinkServiceName, privateEndpointConnectionName string) *retry.Error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DeletePEConnection", ctx, resourceGroupName, privateLinkServiceName, privateEndpointConnectionName)
+	ret0, _ := ret[0].(*retry.Error)
+	return ret0
+}
+
+// DeletePEConnection indicates an expected call of DeletePEConnection.
+func (mr *MockInterfaceMockRecorder) DeletePEConnection(ctx, resourceGroupName, privateLinkServiceName, privateEndpointConnectionName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeletePEConnection", reflect.TypeOf((*MockInterface)(nil).DeletePEConnection), ctx, resourceGroupName, privateLinkServiceName, privateEndpointConnectionName)
+}
+
 // Get mocks base method.
 func (m *MockInterface) Get(ctx context.Context, resourceGroupName, privateLinkServiceName, expand string) (network.PrivateLinkService, *retry.Error) {
 	m.ctrl.T.Helper()
diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go
index 6c3bc9f31..e265a56f2 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -393,3 +393,49 @@ const (
 )
 
 type HealthProbeParams string
+
+// private link service
+const (
+	// ServiceAnnotationPLSCreation determines whether a PLS needs to be created.
+	ServiceAnnotationPLSCreation = "service.beta.kubernetes.io/azure-pls-create"
+
+	// ServiceAnnotationPLSName determines name of the PLS resource to create.
+	ServiceAnnotationPLSName = "service.beta.kubernetes.io/azure-pls-name"
+
+	// ServiceAnnotationPLSIpConfigurationSubnet determines the subnet name to deploy the PLS resource.
+	ServiceAnnotationPLSIpConfigurationSubnet = "service.beta.kubernetes.io/azure-pls-ip-configuration-subnet"
+
+	// ServiceAnnotationPLSIpConfigurationIPAddressCount determines number of IPs to be associated with the PLS.
+	ServiceAnnotationPLSIpConfigurationIPAddressCount = "service.beta.kubernetes.io/azure-pls-ip-configuration-ip-address-count"
+
+	// ServiceAnnotationPLSIPConfigurationIPAddress determines a space separated list of static IPs for the PLS.
+	// Total number of IPs should not be greater than the IP count specified in ServiceAnnotationPLSIpConfigurationIPAddressCount.
+	// If there are fewer IPs specified, the rest are dynamically allocated. The first IP in the list is set as Primary.
+	ServiceAnnotationPLSIpConfigurationIPAddress = "service.beta.kubernetes.io/azure-pls-ip-configuration-ip-address"
+
+	// ServiceAnnotationPLSFqdns determines a space separated list of fqdns associated with the PLS.
+	ServiceAnnotationPLSFqdns = "service.beta.kubernetes.io/azure-pls-fqdns"
+
+	// ServiceAnnotationPLSProxyProtocol determines whether TCP Proxy protocol needs to be enabled on the PLS.
+	ServiceAnnotationPLSProxyProtocol = "service.beta.kubernetes.io/azure-pls-proxy-protocol"
+
+	// ServiceAnnotationPLSVisibility determines a space separated list of Azure subscription IDs for which the PLS is visible.
+	// Use "*" to expose the PLS to all subscriptions.
+	ServiceAnnotationPLSVisibility = "service.beta.kubernetes.io/azure-pls-visibility"
+
+	// ServiceAnnotationPLSAutoApproval determines a space separated list of Azure subscription IDs from which requests can be
+	// automatically approved, only works when visibility is set to "*".
+	ServiceAnnotationPLSAutoApproval = "service.beta.kubernetes.io/azure-pls-auto-approval"
+
+	// ID string used to create a not existing PLS placehold in plsCache to avoid redundant
+	PrivateLinkServiceNotExistID = "PrivateLinkServiceNotExistID"
+
+	// Key of tag indicating owner service of the PLS.
+	OwnerServiceTagKey = "k8s-azure-owner-service"
+
+	// Key of tag indicating cluster name of the service that owns PLS.
+	ClusterNameTagKey = "k8s-azure-cluster-name"
+
+	// Default number of IP configs for PLS
+	PLSDefaultNumOfIPConfig = 1
+)
diff --git a/pkg/provider/azure.go b/pkg/provider/azure.go
index 0be98237c..fcfb29918 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -50,6 +50,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/fileclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/interfaceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/loadbalancerclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatelinkserviceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/publicipclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/routeclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/routetableclient"
@@ -221,6 +222,8 @@ type Config struct {
 	NsgCacheTTLInSeconds int `json:"nsgCacheTTLInSeconds,omitempty" yaml:"nsgCacheTTLInSeconds,omitempty"`
 	// RouteTableCacheTTLInSeconds sets the cache TTL for route table
 	RouteTableCacheTTLInSeconds int `json:"routeTableCacheTTLInSeconds,omitempty" yaml:"routeTableCacheTTLInSeconds,omitempty"`
+	// PlsCacheTTLInSeconds sets the cache TTL for private link service resource
+	PlsCacheTTLInSeconds int `json:"plsCacheTTLInSeconds,omitempty" yaml:"plsCacheTTLInSeconds,omitempty"`
 	// AvailabilitySetsCacheTTLInSeconds sets the cache TTL for VMAS
 	AvailabilitySetsCacheTTLInSeconds int `json:"availabilitySetsCacheTTLInSeconds,omitempty" yaml:"availabilitySetsCacheTTLInSeconds,omitempty"`
 	// PublicIPCacheTTLInSeconds sets the cache TTL for public ip
@@ -231,6 +234,8 @@ type Config struct {
 	// PutVMSSVMBatchSize defines how many requests the client send concurrently when putting the VMSS VMs.
 	// If it is smaller than or equal to zero, the request will be sent one by one in sequence (default).
 	PutVMSSVMBatchSize int `json:"putVMSSVMBatchSize" yaml:"putVMSSVMBatchSize"`
+	// PrivateLinkServiceResourceGroup determines the specific resource group of the private link services user want to use
+	PrivateLinkServiceResourceGroup string `json:"privateLinkServiceResourceGroup,omitempty" yaml:"privateLinkServiceResourceGroup,omitempty"`
 }
 
 type InitSecretConfig struct {
@@ -276,6 +281,7 @@ type Cloud struct {
 	VirtualMachineSizesClient       vmsizeclient.Interface
 	AvailabilitySetsClient          vmasclient.Interface
 	ZoneClient                      zoneclient.Interface
+	PrivateLinkServiceClient        privatelinkserviceclient.Interface
 
 	ResourceRequestBackoff wait.Backoff
 	metadata               *InstanceMetadataService
@@ -320,6 +326,8 @@ type Cloud struct {
 	nsgCache *azcache.TimedCache
 	rtCache  *azcache.TimedCache
 	pipCache *azcache.TimedCache
+	// use LB frontEndIpConfiguration ID as the key and search for PLS attached to the frontEnd
+	plsCache *azcache.TimedCache
 
 	*BlobDiskController
 	*ManagedDiskController
@@ -452,6 +460,10 @@ func (az *Cloud) InitializeCloudFromConfig(config *Config, fromSecret, callFromC
 		config.SecurityGroupResourceGroup = config.ResourceGroup
 	}
 
+	if config.PrivateLinkServiceResourceGroup == "" {
+		config.PrivateLinkServiceResourceGroup = config.ResourceGroup
+	}
+
 	if config.VMType == "" {
 		// default to standard vmType if not set.
 		config.VMType = consts.VMTypeStandard
@@ -608,6 +620,11 @@ func (az *Cloud) initCaches() (err error) {
 		return err
 	}
 
+	az.plsCache, err = az.newPLSCache()
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -762,6 +779,7 @@ func (az *Cloud) configAzureClients(
 	az.PublicIPAddressesClient = publicipclient.New(publicIPClientConfig)
 	az.FileClient = fileclient.New(fileClientConfig)
 	az.AvailabilitySetsClient = vmasclient.New(vmasClientConfig)
+	az.PrivateLinkServiceClient = privatelinkserviceclient.New(azClientConfig)
 
 	if az.ZoneClient == nil {
 		az.ZoneClient = zoneclient.New(zoneClientConfig)
diff --git a/pkg/provider/azure_backoff.go b/pkg/provider/azure_backoff.go
index 79248f3c9..0b8b0b91c 100644
--- a/pkg/provider/azure_backoff.go
+++ b/pkg/provider/azure_backoff.go
@@ -529,3 +529,86 @@ func (az *Cloud) CreateOrUpdateVMSS(resourceGroupName string, VMScaleSetName str
 
 	return nil
 }
+
+func (az *Cloud) CreateOrUpdatePLS(service *v1.Service, pls network.PrivateLinkService) error {
+	ctx, cancel := getContextWithCancel()
+	defer cancel()
+
+	rerr := az.PrivateLinkServiceClient.CreateOrUpdate(ctx, az.PrivateLinkServiceResourceGroup, to.String(pls.Name), pls, to.String(pls.Etag))
+	if rerr == nil {
+		// Invalidate the cache right after updating
+		_ = az.plsCache.Delete(to.String((*pls.LoadBalancerFrontendIPConfigurations)[0].ID))
+		return nil
+	}
+
+	rtJSON, _ := json.Marshal(pls)
+	klog.Warningf("PrivateLinkServiceClient.CreateOrUpdate(%s) failed: %v, PrivateLinkService request: %s", to.String(pls.Name), rerr.Error(), string(rtJSON))
+
+	// Invalidate the cache because etag mismatch.
+	if rerr.HTTPStatusCode == http.StatusPreconditionFailed {
+		klog.V(3).Infof("Private link service cache for %s is cleanup because of http.StatusPreconditionFailed", to.String(pls.Name))
+		_ = az.plsCache.Delete(to.String((*pls.LoadBalancerFrontendIPConfigurations)[0].ID))
+	}
+	// Invalidate the cache because another new operation has canceled the current request.
+	if strings.Contains(strings.ToLower(rerr.Error().Error()), consts.OperationCanceledErrorMessage) {
+		klog.V(3).Infof("Private link service for %s is cleanup because CreateOrUpdatePrivateLinkService is canceled by another operation", to.String(pls.Name))
+		_ = az.plsCache.Delete(to.String((*pls.LoadBalancerFrontendIPConfigurations)[0].ID))
+	}
+	klog.Errorf("PrivateLinkServiceClient.CreateOrUpdate(%s) failed: %v", to.String(pls.Name), rerr.Error())
+	return rerr.Error()
+}
+
+// DeletePLS invokes az.PrivateLinkServiceClient.Delete with exponential backoff retry
+func (az *Cloud) DeletePLS(service *v1.Service, plsName string, plsLBFrontendID string) *retry.Error {
+	ctx, cancel := getContextWithCancel()
+	defer cancel()
+
+	rerr := az.PrivateLinkServiceClient.Delete(ctx, az.PrivateLinkServiceResourceGroup, plsName)
+	if rerr == nil {
+		// Invalidate the cache right after deleting
+		_ = az.plsCache.Delete(plsLBFrontendID)
+		return nil
+	}
+
+	klog.Errorf("PrivateLinkServiceClient.DeletePLS(%s) failed: %s", plsName, rerr.Error().Error())
+	az.Event(service, v1.EventTypeWarning, "DeletePrivateLinkService", rerr.Error().Error())
+	return rerr
+}
+
+// DeletePEConn invokes az.PrivateLinkServiceClient.DeletePEConnection with exponential backoff retry
+func (az *Cloud) DeletePEConn(service *v1.Service, plsName string, peConnName string) *retry.Error {
+	ctx, cancel := getContextWithCancel()
+	defer cancel()
+
+	rerr := az.PrivateLinkServiceClient.DeletePEConnection(ctx, az.PrivateLinkServiceResourceGroup, plsName, peConnName)
+	if rerr == nil {
+		return nil
+	}
+
+	klog.Errorf("PrivateLinkServiceClient.DeletePEConnection(%s-%s) failed: %s", plsName, peConnName, rerr.Error().Error())
+	az.Event(service, v1.EventTypeWarning, "DeletePrivateEndpointConnection", rerr.Error().Error())
+	return rerr
+}
+
+// CreateOrUpdateSubnet invokes az.SubnetClient.CreateOrUpdate with exponential backoff retry
+func (az *Cloud) CreateOrUpdateSubnet(service *v1.Service, subnet network.Subnet) error {
+	ctx, cancel := getContextWithCancel()
+	defer cancel()
+
+	var rg string
+	if len(az.VnetResourceGroup) > 0 {
+		rg = az.VnetResourceGroup
+	} else {
+		rg = az.ResourceGroup
+	}
+
+	rerr := az.SubnetsClient.CreateOrUpdate(ctx, rg, az.VnetName, *subnet.Name, subnet)
+	klog.V(10).Infof("SubnetClient.CreateOrUpdate(%s): end", *subnet.Name)
+	if rerr != nil {
+		klog.Errorf("SubnetClient.CreateOrUpdate(%s) failed: %s", *subnet.Name, rerr.Error().Error())
+		az.Event(service, v1.EventTypeWarning, "CreateOrUpdateSubnet", rerr.Error().Error())
+		return rerr.Error()
+	}
+
+	return nil
+}
diff --git a/pkg/provider/azure_fakes.go b/pkg/provider/azure_fakes.go
index 8c4e620a8..98a6aac53 100644
--- a/pkg/provider/azure_fakes.go
+++ b/pkg/provider/azure_fakes.go
@@ -28,6 +28,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/diskclient/mockdiskclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/interfaceclient/mockinterfaceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/loadbalancerclient/mockloadbalancerclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/publicipclient/mockpublicipclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/routeclient/mockrouteclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/routetableclient/mockroutetableclient"
@@ -66,19 +67,20 @@ func GetTestCloud(ctrl *gomock.Controller) (az *Cloud) {
 				TenantID:       "tenant",
 				SubscriptionID: "subscription",
 			},
-			ResourceGroup:                "rg",
-			VnetResourceGroup:            "rg",
-			RouteTableResourceGroup:      "rg",
-			SecurityGroupResourceGroup:   "rg",
-			Location:                     "westus",
-			VnetName:                     "vnet",
-			SubnetName:                   "subnet",
-			SecurityGroupName:            "nsg",
-			RouteTableName:               "rt",
-			PrimaryAvailabilitySetName:   "as",
-			PrimaryScaleSetName:          "vmss",
-			MaximumLoadBalancerRuleCount: 250,
-			VMType:                       consts.VMTypeStandard,
+			ResourceGroup:                   "rg",
+			VnetResourceGroup:               "rg",
+			RouteTableResourceGroup:         "rg",
+			SecurityGroupResourceGroup:      "rg",
+			PrivateLinkServiceResourceGroup: "rg",
+			Location:                        "westus",
+			VnetName:                        "vnet",
+			SubnetName:                      "subnet",
+			SecurityGroupName:               "nsg",
+			RouteTableName:                  "rt",
+			PrimaryAvailabilitySetName:      "as",
+			PrimaryScaleSetName:             "vmss",
+			MaximumLoadBalancerRuleCount:    250,
+			VMType:                          consts.VMTypeStandard,
 		},
 		nodeZones:                map[string]sets.String{},
 		nodeInformerSynced:       func() bool { return true },
@@ -99,12 +101,14 @@ func GetTestCloud(ctrl *gomock.Controller) (az *Cloud) {
 	az.VirtualMachineScaleSetsClient = mockvmssclient.NewMockInterface(ctrl)
 	az.VirtualMachineScaleSetVMsClient = mockvmssvmclient.NewMockInterface(ctrl)
 	az.VirtualMachinesClient = mockvmclient.NewMockInterface(ctrl)
+	az.PrivateLinkServiceClient = mockprivatelinkserviceclient.NewMockInterface(ctrl)
 	az.VMSet, _ = newAvailabilitySet(az)
 	az.vmCache, _ = az.newVMCache()
 	az.lbCache, _ = az.newLBCache()
 	az.nsgCache, _ = az.newNSGCache()
 	az.rtCache, _ = az.newRouteTableCache()
 	az.pipCache, _ = az.newPIPCache()
+	az.plsCache, _ = az.newPLSCache()
 
 	common := &controllerCommon{cloud: az, resourceGroup: "rg", location: "westus"}
 	az.controllerCommon = common
diff --git a/pkg/provider/azure_loadbalancer.go b/pkg/provider/azure_loadbalancer.go
index 78d86309c..69e440dda 100644
--- a/pkg/provider/azure_loadbalancer.go
+++ b/pkg/provider/azure_loadbalancer.go
@@ -95,7 +95,7 @@ func (az *Cloud) reconcileService(ctx context.Context, clusterName string, servi
 		return nil, err
 	}
 
-	lbStatus, _, err := az.getServiceLoadBalancerStatus(service, lb, nil)
+	lbStatus, fipConfig, err := az.getServiceLoadBalancerStatus(service, lb, nil)
 	if err != nil {
 		klog.Errorf("getServiceLoadBalancerStatus(%s) failed: %v", serviceName, err)
 		return nil, err
@@ -111,6 +111,13 @@ func (az *Cloud) reconcileService(ctx context.Context, clusterName string, servi
 		return nil, err
 	}
 
+	if fipConfig != nil {
+		if err := az.reconcilePrivateLinkService(clusterName, service, fipConfig, true /* wantPLS */); err != nil {
+			klog.Errorf("reconcilePrivateLinkService(%s) failed: %#v", serviceName, err)
+			return nil, err
+		}
+	}
+
 	updateService := updateServiceLoadBalancerIP(service, to.String(serviceIP))
 	flippedService := flipServiceInternalAnnotation(updateService)
 	if _, err := az.reconcileLoadBalancer(clusterName, flippedService, nil, false /* wantLb */); err != nil {
@@ -386,6 +393,13 @@ func (az *Cloud) removeFrontendIPConfigurationFromLoadBalancer(lb *network.LoadB
 		lb.Probes = &lbProbes
 	}
 
+	// clean up any private link service associated with the frontEndIPConfig
+	err := az.reconcilePrivateLinkService(clusterName, service, fip, false /* wantPLS */)
+	if err != nil {
+		klog.Errorf("removeFrontendIPConfigurationFromLoadBalancer(%s, %s, %s, %s): failed to clean up PLS: %v", to.String(lb.Name), to.String(fip.Name), clusterName, service.Name, err)
+		return err
+	}
+
 	if len(fipConfigs) == 0 {
 		klog.V(2).Infof("removeFrontendIPConfigurationFromLoadBalancer(%s, %s, %s, %s): deleting load balancer because there is no remaining frontend IP configurations", to.String(lb.Name), to.String(fip.Name), clusterName, service.Name)
 		err := az.cleanOrphanedLoadBalancer(lb, existingLBs, service, clusterName)
@@ -1507,7 +1521,7 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 	}
 
 	// reconcile the load balancer's frontend IP configurations.
-	ownedFIPConfig, changed, err := az.reconcileFrontendIPConfigs(clusterName, service, lb, lbStatus, wantLb, defaultLBFrontendIPConfigName)
+	ownedFIPConfig, toDeleteConfigs, changed, err := az.reconcileFrontendIPConfigs(clusterName, service, lb, lbStatus, wantLb, defaultLBFrontendIPConfigName)
 	if err != nil {
 		return lb, err
 	}
@@ -1559,6 +1573,22 @@ func (az *Cloud) reconcileLoadBalancer(clusterName string, service *v1.Service,
 	// We only care about if there is any change in the LB, which means dirtyLB
 	// If it is not exist, and no change to that, we don't CreateOrUpdate LB
 	if dirtyLb {
+		if len(toDeleteConfigs) > 0 {
+			for i := range toDeleteConfigs {
+				fipConfigToDel := toDeleteConfigs[i]
+				err := az.reconcilePrivateLinkService(clusterName, service, &fipConfigToDel, false /* wantPLS */)
+				if err != nil {
+					klog.Errorf(
+						"reconcileLoadBalancer for service(%s): lb(%s) - failed to clean up PrivateLinkService for frontEnd(%s): %v",
+						serviceName,
+						lbName,
+						to.String(fipConfigToDel.Name),
+						err,
+					)
+				}
+			}
+		}
+
 		if lb.FrontendIPConfigurations == nil || len(*lb.FrontendIPConfigurations) == 0 {
 			err := az.cleanOrphanedLoadBalancer(lb, existingLBs, service, clusterName)
 			if err != nil {
@@ -1695,13 +1725,14 @@ func (az *Cloud) reconcileLBRules(lb *network.LoadBalancer, service *v1.Service,
 	return dirtyRules
 }
 
-func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Service, lb *network.LoadBalancer, status *v1.LoadBalancerStatus, wantLb bool, defaultLBFrontendIPConfigName string) (*network.FrontendIPConfiguration, bool, error) {
+func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Service, lb *network.LoadBalancer, status *v1.LoadBalancerStatus, wantLb bool, defaultLBFrontendIPConfigName string) (*network.FrontendIPConfiguration, []network.FrontendIPConfiguration, bool, error) {
 	var err error
 	lbName := *lb.Name
 	serviceName := getServiceName(service)
 	isInternal := requiresInternalLoadBalancer(service)
 	dirtyConfigs := false
 	var newConfigs []network.FrontendIPConfiguration
+	var toDeleteConfigs []network.FrontendIPConfiguration
 	if lb.FrontendIPConfigurations != nil {
 		newConfigs = *lb.FrontendIPConfigurations
 	}
@@ -1714,12 +1745,12 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 			config := newConfigs[i]
 			isServiceOwnsFrontendIP, _, err := az.serviceOwnsFrontendIP(config, service, pips)
 			if err != nil {
-				return nil, false, err
+				return nil, toDeleteConfigs, false, err
 			}
 			if isServiceOwnsFrontendIP {
 				unsafe, err := az.isFrontendIPConfigUnsafeToDelete(lb, service, config.ID)
 				if err != nil {
-					return nil, false, err
+					return nil, toDeleteConfigs, false, err
 				}
 
 				// If the frontend IP configuration is not being referenced by:
@@ -1737,6 +1768,7 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 						klog.V(2).Infof("reconcileLoadBalancer for service (%s)(%t): nil name of lb frontendconfig", serviceName, wantLb)
 					}
 
+					toDeleteConfigs = append(toDeleteConfigs, newConfigs[i])
 					newConfigs = append(newConfigs[:i], newConfigs[i+1:]...)
 					dirtyConfigs = true
 				}
@@ -1757,10 +1789,11 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 			klog.V(4).Infof("reconcileFrontendIPConfigs for service (%s): checking owned frontend IP cofiguration %s", serviceName, to.String(config.Name))
 			isFipChanged, err = az.isFrontendIPChanged(clusterName, config, service, defaultLBFrontendIPConfigName, pips)
 			if err != nil {
-				return nil, false, err
+				return nil, toDeleteConfigs, false, err
 			}
 			if isFipChanged {
 				klog.V(2).Infof("reconcileLoadBalancer for service (%s)(%t): lb frontendconfig(%s) - dropping", serviceName, wantLb, *config.Name)
+				toDeleteConfigs = append(toDeleteConfigs, newConfigs[i])
 				newConfigs = append(newConfigs[:i], newConfigs[i+1:]...)
 				dirtyConfigs = true
 				previousZone = config.Zones
@@ -1770,7 +1803,7 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 
 		ownedFIPConfig, _, err = az.findFrontendIPConfigOfService(&newConfigs, service, pips)
 		if err != nil {
-			return nil, false, err
+			return nil, toDeleteConfigs, false, err
 		}
 
 		if ownedFIPConfig == nil {
@@ -1785,11 +1818,11 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 				}
 				subnet, existsSubnet, err := az.getSubnet(az.VnetName, *subnetName)
 				if err != nil {
-					return nil, false, err
+					return nil, toDeleteConfigs, false, err
 				}
 
 				if !existsSubnet {
-					return nil, false, fmt.Errorf("ensure(%s): lb(%s) - failed to get subnet: %s/%s", serviceName, lbName, az.VnetName, az.SubnetName)
+					return nil, toDeleteConfigs, false, fmt.Errorf("ensure(%s): lb(%s) - failed to get subnet: %s/%s", serviceName, lbName, az.VnetName, az.SubnetName)
 				}
 
 				configProperties := network.FrontendIPConfigurationPropertiesFormat{
@@ -1817,12 +1850,12 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 			} else {
 				pipName, shouldPIPExisted, err := az.determinePublicIPName(clusterName, service, pips)
 				if err != nil {
-					return nil, false, err
+					return nil, toDeleteConfigs, false, err
 				}
 				domainNameLabel, found := getPublicIPDomainNameLabel(service)
 				pip, err := az.ensurePublicIPExists(service, pipName, domainNameLabel, clusterName, shouldPIPExisted, found)
 				if err != nil {
-					return nil, false, err
+					return nil, toDeleteConfigs, false, err
 				}
 				fipConfigurationProperties = &network.FrontendIPConfigurationPropertiesFormat{
 					PublicIPAddress: &network.PublicIPAddress{ID: pip.ID},
@@ -1838,7 +1871,7 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 			if isInternal {
 				if err := az.getFrontendZones(&newConfig, previousZone, isFipChanged, serviceName, defaultLBFrontendIPConfigName); err != nil {
 					klog.Errorf("reconcileLoadBalancer for service (%s)(%t): failed to getFrontendZones: %s", serviceName, wantLb, err.Error())
-					return nil, false, err
+					return nil, toDeleteConfigs, false, err
 				}
 			}
 			newConfigs = append(newConfigs, newConfig)
@@ -1851,7 +1884,7 @@ func (az *Cloud) reconcileFrontendIPConfigs(clusterName string, service *v1.Serv
 		lb.FrontendIPConfigurations = &newConfigs
 	}
 
-	return ownedFIPConfig, dirtyConfigs, err
+	return ownedFIPConfig, toDeleteConfigs, dirtyConfigs, err
 }
 
 func (az *Cloud) getFrontendZones(
diff --git a/pkg/provider/azure_privatelinkservice.go b/pkg/provider/azure_privatelinkservice.go
new file mode 100644
index 000000000..fe7604010
--- /dev/null
+++ b/pkg/provider/azure_privatelinkservice.go
@@ -0,0 +1,670 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provider
+
+import (
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
+	"github.com/Azure/go-autorest/autorest/to"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
+
+	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
+	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+// reconcilePrivateLinkService() function makes sure a PLS is created or deleted on
+// a Load Balancer frontend IP Configuration according to service spec and cluster operation
+func (az *Cloud) reconcilePrivateLinkService(
+	clusterName string,
+	service *v1.Service,
+	fipConfig *network.FrontendIPConfiguration,
+	wantPLS bool,
+) error {
+	createPLS := wantPLS && serviceRequiresPLS(service)
+	serviceName := getServiceName(service)
+	fipConfigID := fipConfig.ID
+	klog.V(2).Infof("reconcilePrivateLinkService for service(%s) - LB fipConfigID(%s) - wantPLS(%t) - createPLS(%t)", serviceName, to.String(fipConfig.Name), wantPLS, createPLS)
+
+	if createPLS {
+		// Firstly, make sure it's internal service
+		if !requiresInternalLoadBalancer(service) {
+			return fmt.Errorf("reconcilePrivateLinkService for service(%s): service requiring private link service must be internal", serviceName)
+		}
+
+		// Secondly, check if there is a private link service already created
+		existingPLS, err := az.getPrivateLinkService(fipConfigID, azcache.CacheReadTypeDefault)
+		if err != nil {
+			klog.Errorf("reconcilePrivateLinkService for service(%s): getPrivateLinkService(%s) failed: %v", serviceName, to.String(fipConfigID), err)
+			return err
+		}
+
+		exists := !strings.EqualFold(to.String(existingPLS.ID), consts.PrivateLinkServiceNotExistID)
+		if exists {
+			klog.V(4).Infof("reconcilePrivateLinkService for service(%s): found existing private link service attached(%s)", serviceName, to.String(existingPLS.Name))
+			if !isManagedPrivateLinkSerivce(&existingPLS, clusterName) {
+				return fmt.Errorf(
+					"reconcilePrivateLinkService for service(%s) failed: LB frontend(%s) already has unmanaged private link service(%s)",
+					serviceName,
+					to.String(fipConfigID),
+					to.String(existingPLS.ID),
+				)
+			}
+			// If there is an existing private link service, only owner service can update its properties
+			ownerService := getPrivateLinkServiceOwner(&existingPLS)
+			if !strings.EqualFold(ownerService, serviceName) {
+				if serviceHasAdditionalConfigs(service) {
+					return fmt.Errorf(
+						"reconcilePrivateLinkService for service(%s) failed: LB frontend(%s) already has existing private link service(%s) owned by service(%s)",
+						serviceName,
+						to.String(fipConfigID),
+						to.String(existingPLS.Name),
+						ownerService,
+					)
+				}
+				klog.V(2).Infof(
+					"reconcilePrivateLinkService for service(%s): automatically share private link service(%s) owned by service(%s)",
+					serviceName,
+					to.String(existingPLS.Name),
+					ownerService,
+				)
+				return nil
+			}
+		} else {
+			existingPLS.ID = nil
+			existingPLS.Location = &az.Location
+			existingPLS.PrivateLinkServiceProperties = &network.PrivateLinkServiceProperties{}
+			if az.HasExtendedLocation() {
+				existingPLS.ExtendedLocation = &network.ExtendedLocation{
+					Name: &az.ExtendedLocationName,
+					Type: getExtendedLocationTypeFromString(az.ExtendedLocationType),
+				}
+			}
+		}
+
+		plsName, err := az.getPrivateLinkServiceName(&existingPLS, service, fipConfig)
+		if err != nil {
+			return err
+		}
+
+		dirtyPLS, err := az.getExpectedPrivateLinkService(&existingPLS, &plsName, &clusterName, service, fipConfig)
+		if err != nil {
+			return err
+		}
+
+		if dirtyPLS {
+			klog.V(2).Infof("reconcilePrivateLinkService for service(%s): pls(%s) - updating", serviceName, plsName)
+			err := az.disablePLSNetworkPolicy(service)
+			if err != nil {
+				klog.Errorf("reconcilePrivateLinkService for service(%s) disable PLS network policy failed for pls(%s): %v", serviceName, plsName, err.Error())
+				return err
+			}
+			existingPLS.Etag = to.StringPtr("")
+			err = az.CreateOrUpdatePLS(service, existingPLS)
+			if err != nil {
+				klog.Errorf("reconcilePrivateLinkService for service(%s) abort backoff: pls(%s) - updating: %s", serviceName, plsName, err.Error())
+				return err
+			}
+		}
+	} else if !wantPLS {
+		existingPLS, err := az.getPrivateLinkService(fipConfigID, azcache.CacheReadTypeDefault)
+		if err != nil {
+			klog.Errorf("reconcilePrivateLinkService for service(%s): getPrivateLinkService(%s) failed: %v", serviceName, to.String(fipConfigID), err)
+			return err
+		}
+
+		exists := !strings.EqualFold(to.String(existingPLS.ID), consts.PrivateLinkServiceNotExistID)
+		if exists {
+			deleteErr := az.safeDeletePLS(&existingPLS, service)
+			if deleteErr != nil {
+				klog.Errorf("reconcilePrivateLinkService for service(%s): deletePLS for frontEnd(%s) failed: %v", serviceName, to.String(fipConfigID), err)
+				return deleteErr.Error()
+			}
+		}
+	}
+
+	klog.V(2).Infof("reconcilePrivateLinkService for service(%s) finished", serviceName)
+	return nil
+}
+
+func (az *Cloud) disablePLSNetworkPolicy(service *v1.Service) error {
+	serviceName := getServiceName(service)
+	subnetName := getPLSSubnetName(service)
+	if subnetName == nil {
+		subnetName = &az.SubnetName
+	}
+
+	subnet, existsSubnet, err := az.getSubnet(az.VnetName, *subnetName)
+	if err != nil {
+		return err
+	}
+	if !existsSubnet {
+		return fmt.Errorf("disablePLSNetworkPolicy: failed to get private link service subnet(%s) for service(%s)", *subnetName, serviceName)
+	}
+
+	// Policy already disabled
+	if subnet.PrivateLinkServiceNetworkPolicies == network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled {
+		return nil
+	}
+
+	subnet.PrivateLinkServiceNetworkPolicies = network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled
+	err = az.CreateOrUpdateSubnet(service, subnet)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (az *Cloud) safeDeletePLS(pls *network.PrivateLinkService, service *v1.Service) *retry.Error {
+	if pls == nil {
+		return nil
+	}
+
+	peConns := pls.PrivateEndpointConnections
+	if peConns != nil {
+		for _, peConn := range *peConns {
+			klog.V(2).Infof("deletePLS: deleting PEConnection %s", to.String(peConn.Name))
+			rerr := az.DeletePEConn(service, to.String(pls.Name), to.String(peConn.Name))
+			if rerr != nil {
+				return rerr
+			}
+		}
+	}
+
+	rerr := az.DeletePLS(service, to.String(pls.Name), to.String((*pls.LoadBalancerFrontendIPConfigurations)[0].ID))
+	if rerr != nil {
+		return rerr
+	}
+	klog.V(2).Infof("safeDeletePLS(%s) finished", to.String(pls.Name))
+	return nil
+}
+
+// getPrivateLinkServiceName() returns the name of private link service, or any error
+func (az *Cloud) getPrivateLinkServiceName(
+	existingPLS *network.PrivateLinkService,
+	service *v1.Service,
+	fipConfig *network.FrontendIPConfiguration,
+) (string, error) {
+	existingName := existingPLS.Name
+	serviceName := getServiceName(service)
+
+	if nameFromService, found := service.Annotations[consts.ServiceAnnotationPLSName]; found {
+		nameFromService = strings.TrimSpace(nameFromService)
+		if existingName != nil && !strings.EqualFold(to.String(existingName), nameFromService) {
+			return "", fmt.Errorf(
+				"getPrivateLinkServiceName(%s) failed: cannot change existing private link service name (%s) to (%s)",
+				serviceName,
+				to.String(existingName),
+				nameFromService,
+			)
+		}
+		return nameFromService, nil
+	}
+
+	if existingName != nil {
+		return to.String(existingName), nil
+	}
+
+	// default PLS name: pls-<frontendIPConfigName>
+	return fmt.Sprintf("%s-%s", "pls", *fipConfig.Name), nil
+}
+
+// getExpectedPrivateLinkService builds expected PLS object from service spec
+func (az *Cloud) getExpectedPrivateLinkService(
+	existingPLS *network.PrivateLinkService,
+	plsName *string,
+	clusterName *string,
+	service *v1.Service,
+	fipConfig *network.FrontendIPConfiguration,
+) (dirtyPLS bool, err error) {
+	dirtyPLS = false
+
+	if existingPLS == nil {
+		return false, fmt.Errorf("getExpectedPrivateLinkService: existingPLS is nil (unexpected)")
+	}
+
+	// This only happens for an empty
+	if existingPLS.Name == nil || !strings.EqualFold(*existingPLS.Name, *plsName) {
+		existingPLS.Name = plsName
+		dirtyPLS = true
+	}
+
+	// Set failed PLS as dirty so that provision can be retried
+	if existingPLS.ProvisioningState == network.ProvisioningStateFailed {
+		dirtyPLS = true
+	}
+
+	// Shadow copy properties to avoid changing pls cache
+	plsProperties := *existingPLS.PrivateLinkServiceProperties
+	existingPLS.PrivateLinkServiceProperties = &plsProperties
+
+	// Set LBFrontendIpConfiguration
+	if existingPLS.LoadBalancerFrontendIPConfigurations == nil {
+		existingPLS.LoadBalancerFrontendIPConfigurations = &[]network.FrontendIPConfiguration{{ID: fipConfig.ID}}
+		dirtyPLS = true
+	}
+
+	changed, err := az.reconcilePLSIpConfigs(existingPLS, service)
+	if err != nil {
+		return false, err
+	}
+	if changed {
+		dirtyPLS = true
+	}
+
+	if reconcilePLSEnableProxyProtocol(existingPLS, service) {
+		dirtyPLS = true
+	}
+
+	if reconcilePLSFqdn(existingPLS, service) {
+		dirtyPLS = true
+	}
+
+	changed, err = reconcilePLSVisibility(existingPLS, service)
+	if err != nil {
+		return false, err
+	}
+	if changed {
+		dirtyPLS = true
+	}
+
+	if az.reconcilePLSTags(existingPLS, clusterName, service) {
+		dirtyPLS = true
+	}
+
+	return dirtyPLS, nil
+}
+
+// reconcile Private link service's IP configurations
+func (az *Cloud) reconcilePLSIpConfigs(
+	existingPLS *network.PrivateLinkService,
+	service *v1.Service,
+) (bool, error) {
+	changed := false
+	serviceName := getServiceName(service)
+
+	subnetName := getPLSSubnetName(service)
+	if subnetName == nil {
+		subnetName = &az.SubnetName
+	}
+	subnet, existsSubnet, err := az.getSubnet(az.VnetName, *subnetName)
+	if err != nil {
+		return false, err
+	}
+	if !existsSubnet {
+		return false, fmt.Errorf("checkAndUpdatePLSIPConfigs: failed to get private link service subnet(%s) for service(%s)", *subnetName, serviceName)
+	}
+
+	ipConfigCount, err := getPLSIPConfigCount(service)
+	if err != nil {
+		return false, err
+	}
+
+	staticIps, primaryIP, err := getPLSStaticIPs(service)
+	if err != nil {
+		return false, err
+	}
+
+	if int(ipConfigCount) < len(staticIps) {
+		return false, fmt.Errorf("checkAndUpdatePLSIPConfigs: ipConfigCount(%d) must be no smaller than number of static IPs specified(%d)", ipConfigCount, len(staticIps))
+	}
+
+	if existingPLS.IPConfigurations == nil {
+		existingPLS.IPConfigurations = &[]network.PrivateLinkServiceIPConfiguration{}
+		changed = true
+	}
+
+	if int32(len(*existingPLS.IPConfigurations)) != ipConfigCount {
+		changed = true
+	}
+
+	existingStaticIps := make([]string, 0)
+	for _, ipConfig := range *existingPLS.IPConfigurations {
+		if !strings.EqualFold(to.String(subnet.ID), to.String(ipConfig.Subnet.ID)) {
+			changed = true
+		}
+		if strings.EqualFold(string(ipConfig.PrivateIPAllocationMethod), string(network.IPAllocationMethodStatic)) {
+			klog.V(10).Infof("Found static IP: %s", to.String(ipConfig.PrivateIPAddress))
+			if _, found := staticIps[to.String(ipConfig.PrivateIPAddress)]; !found {
+				changed = true
+			}
+			existingStaticIps = append(existingStaticIps, to.String(ipConfig.PrivateIPAddress))
+		}
+		if *ipConfig.Primary {
+			if strings.EqualFold(string(ipConfig.PrivateIPAllocationMethod), string(network.IPAllocationMethodStatic)) {
+				if !strings.EqualFold(primaryIP, to.String(ipConfig.PrivateIPAddress)) {
+					changed = true
+				}
+			} else {
+				// Dynamic
+				if primaryIP != "" {
+					changed = true
+				}
+			}
+		}
+	}
+	if len(existingStaticIps) != len(staticIps) {
+		changed = true
+	}
+
+	if changed {
+		ipConfigs := []network.PrivateLinkServiceIPConfiguration{}
+		for k := range staticIps {
+			ip := k
+			isPrimary := strings.EqualFold(ip, primaryIP)
+			configName := fmt.Sprintf("%s-%s-static-%s", to.String(subnet.Name), to.String(existingPLS.Name), ip)
+			ipConfigs = append(ipConfigs, network.PrivateLinkServiceIPConfiguration{
+				Name: &configName,
+				PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+					PrivateIPAddress:          &ip,
+					PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+					Subnet: &network.Subnet{
+						ID: subnet.ID,
+					},
+					Primary:                 &isPrimary,
+					PrivateIPAddressVersion: network.IPVersionIPv4,
+				},
+			})
+		}
+		for i := 0; i < int(ipConfigCount)-len(staticIps); i++ {
+			isPrimary := primaryIP == "" && i == 0
+			configName := fmt.Sprintf("%s-%s-dynamic-%d", to.String(subnet.Name), to.String(existingPLS.Name), i)
+			ipConfigs = append(ipConfigs, network.PrivateLinkServiceIPConfiguration{
+				Name: &configName,
+				PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+					PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+					Subnet: &network.Subnet{
+						ID: subnet.ID,
+					},
+					Primary:                 &isPrimary,
+					PrivateIPAddressVersion: network.IPVersionIPv4,
+				},
+			})
+		}
+		existingPLS.IPConfigurations = &ipConfigs
+	}
+	return changed, nil
+}
+
+func serviceRequiresPLS(service *v1.Service) bool {
+	return getBoolValueFromServiceAnnotations(service, consts.ServiceAnnotationPLSCreation)
+}
+
+func reconcilePLSEnableProxyProtocol(
+	existingPLS *network.PrivateLinkService,
+	service *v1.Service,
+) bool {
+	changed := false
+	enableProxyProtocol := getBoolValueFromServiceAnnotations(service, consts.ServiceAnnotationPLSProxyProtocol)
+	if enableProxyProtocol && (existingPLS.EnableProxyProtocol == nil || !*existingPLS.EnableProxyProtocol) {
+		changed = true
+	} else if !enableProxyProtocol && (existingPLS.EnableProxyProtocol != nil && *existingPLS.EnableProxyProtocol) {
+		changed = true
+	}
+	if changed {
+		existingPLS.EnableProxyProtocol = &enableProxyProtocol
+	}
+	return changed
+}
+
+func reconcilePLSFqdn(
+	existingPLS *network.PrivateLinkService,
+	service *v1.Service,
+) bool {
+	changed := false
+	fqdns := getPLSFqdns(service)
+	if existingPLS.Fqdns == nil {
+		if len(fqdns) != 0 {
+			changed = true
+		}
+	} else if !sameContentInSlices(fqdns, *existingPLS.Fqdns) {
+		changed = true
+	}
+
+	if changed {
+		existingPLS.Fqdns = &fqdns
+	}
+	return changed
+}
+
+func reconcilePLSVisibility(
+	existingPLS *network.PrivateLinkService,
+	service *v1.Service,
+) (bool, error) {
+	changed := false
+	visibilitySubs, anyoneCanView := getPLSVisibility(service)
+	autoApprovalSubs := getPLSAutoApproval(service)
+	if !anyoneCanView && len(autoApprovalSubs) > 0 {
+		return false, fmt.Errorf("reconcilePLSVisibility: autoApproval only takes effect when visibility is set to \"*\"")
+	}
+
+	if existingPLS.Visibility == nil || existingPLS.Visibility.Subscriptions == nil {
+		if len(visibilitySubs) != 0 {
+			changed = true
+		}
+	} else if !sameContentInSlices(visibilitySubs, *existingPLS.Visibility.Subscriptions) {
+		changed = true
+	}
+
+	if existingPLS.AutoApproval == nil || existingPLS.AutoApproval.Subscriptions == nil {
+		if len(autoApprovalSubs) != 0 {
+			changed = true
+		}
+	} else if !sameContentInSlices(autoApprovalSubs, *existingPLS.AutoApproval.Subscriptions) {
+		changed = true
+	}
+
+	if changed {
+		existingPLS.Visibility = &network.PrivateLinkServicePropertiesVisibility{
+			Subscriptions: &visibilitySubs,
+		}
+		existingPLS.AutoApproval = &network.PrivateLinkServicePropertiesAutoApproval{
+			Subscriptions: &autoApprovalSubs,
+		}
+	}
+	return changed, nil
+}
+
+func (az *Cloud) reconcilePLSTags(
+	existingPLS *network.PrivateLinkService,
+	clusterName *string,
+	service *v1.Service,
+) bool {
+	configTags := parseTags(az.Tags, az.TagsMap)
+	serviceName := getServiceName(service)
+
+	if existingPLS.Tags == nil {
+		existingPLS.Tags = make(map[string]*string)
+	}
+
+	tags := existingPLS.Tags
+
+	// include the cluster name and service name tags when comparing
+	if v, ok := tags[consts.ClusterNameTagKey]; ok && v != nil {
+		configTags[consts.ClusterNameTagKey] = v
+	} else {
+		configTags[consts.ClusterNameTagKey] = clusterName
+	}
+	if v, ok := tags[consts.OwnerServiceTagKey]; ok && v != nil {
+		configTags[consts.OwnerServiceTagKey] = v
+	} else {
+		configTags[consts.OwnerServiceTagKey] = &serviceName
+	}
+
+	tags, changed := az.reconcileTags(existingPLS.Tags, configTags)
+	existingPLS.Tags = tags
+
+	return changed
+}
+
+func getPLSSubnetName(service *v1.Service) *string {
+	if l, found := service.Annotations[consts.ServiceAnnotationPLSIpConfigurationSubnet]; found && strings.TrimSpace(l) != "" {
+		return &l
+	}
+
+	if l, found := service.Annotations[consts.ServiceAnnotationLoadBalancerInternalSubnet]; found && strings.TrimSpace(l) != "" {
+		return &l
+	}
+
+	return nil
+}
+
+func getPLSIPConfigCount(service *v1.Service) (int32, error) {
+	ipConfigCnt, err := consts.Getint32ValueFromK8sSvcAnnotation(
+		service.Annotations,
+		consts.ServiceAnnotationPLSIpConfigurationIPAddressCount,
+		func(val *int32) error {
+			const (
+				MinimumNumOfIPConfig = 1
+				MaximumNumOfIPConfig = 8
+			)
+			if *val < MinimumNumOfIPConfig {
+				return fmt.Errorf("minimum number of private link service ipConfig is %d, %d provided", MinimumNumOfIPConfig, *val)
+			}
+			if *val > MaximumNumOfIPConfig {
+				return fmt.Errorf("maximum number of private link service ipConfig is %d, %d provided", MaximumNumOfIPConfig, *val)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return 0, err
+	}
+	if ipConfigCnt != nil {
+		return *ipConfigCnt, nil
+	}
+	return consts.PLSDefaultNumOfIPConfig, nil
+}
+
+func getPLSFqdns(service *v1.Service) []string {
+	fqdns := make([]string, 0)
+	if v, ok := service.Annotations[consts.ServiceAnnotationPLSFqdns]; ok {
+		fqdnList := strings.Split(strings.TrimSpace(v), " ")
+		for _, fqdn := range fqdnList {
+			fqdn = strings.TrimSpace(fqdn)
+			if fqdn == "" {
+				continue
+			}
+			fqdns = append(fqdns, fqdn)
+		}
+	}
+	return fqdns
+}
+
+func getPLSVisibility(service *v1.Service) ([]string, bool) {
+	visibilityList := make([]string, 0)
+	if val, ok := service.Annotations[consts.ServiceAnnotationPLSVisibility]; ok {
+		visibilities := strings.Split(strings.TrimSpace(val), " ")
+		for _, vis := range visibilities {
+			vis = strings.TrimSpace(vis)
+			if vis == "" {
+				continue
+			}
+			if vis == "*" {
+				visibilityList = []string{"*"}
+				return visibilityList, true
+			}
+			visibilityList = append(visibilityList, vis)
+		}
+	}
+	return visibilityList, false
+}
+
+func getPLSAutoApproval(service *v1.Service) []string {
+	autoApprovalList := make([]string, 0)
+	if val, ok := service.Annotations[consts.ServiceAnnotationPLSAutoApproval]; ok {
+		autoApprovals := strings.Split(strings.TrimSpace(val), " ")
+		for _, autoApp := range autoApprovals {
+			autoApp = strings.TrimSpace(autoApp)
+			if autoApp == "" {
+				continue
+			}
+			autoApprovalList = append(autoApprovalList, autoApp)
+		}
+	}
+	return autoApprovalList
+}
+
+func getPLSStaticIPs(service *v1.Service) (map[string]bool, string, error) {
+	result := make(map[string]bool)
+	primaryIP := ""
+	if val, ok := service.Annotations[consts.ServiceAnnotationPLSIpConfigurationIPAddress]; ok {
+		ips := strings.Split(strings.TrimSpace(val), " ")
+		for _, ip := range ips {
+			ip = strings.TrimSpace(ip)
+			if ip == "" {
+				continue // skip empty string
+			}
+
+			parsedIP := net.ParseIP(ip)
+			if parsedIP == nil {
+				return nil, "", fmt.Errorf("getPLSStaticIPs: %s is not a valid IP address", ip)
+			}
+
+			if parsedIP.To4() == nil {
+				return nil, "", fmt.Errorf("getPLSStaticIPs: private link service ip config only supports IPv4, %s provided", ip)
+			}
+
+			result[ip] = true
+			if primaryIP == "" {
+				primaryIP = ip
+			}
+		}
+	}
+
+	return result, primaryIP, nil
+}
+
+func isManagedPrivateLinkSerivce(existingPLS *network.PrivateLinkService, clusterName string) bool {
+	tags := existingPLS.Tags
+	v, ok := tags[consts.ClusterNameTagKey]
+	return ok && v != nil && strings.EqualFold(strings.TrimSpace(*v), clusterName)
+}
+
+// find owner service for an existing private link service from its tags
+func getPrivateLinkServiceOwner(existingPLS *network.PrivateLinkService) string {
+	tags := existingPLS.Tags
+	v, ok := tags[consts.OwnerServiceTagKey]
+	if ok && v != nil {
+		return *v
+	}
+	return ""
+}
+
+// Return true if service has private link service config annotations
+func serviceHasAdditionalConfigs(service *v1.Service) bool {
+	tagKeyList := []string{
+		consts.ServiceAnnotationPLSName,
+		consts.ServiceAnnotationPLSIpConfigurationSubnet,
+		consts.ServiceAnnotationPLSIpConfigurationIPAddressCount,
+		consts.ServiceAnnotationPLSIpConfigurationIPAddress,
+		consts.ServiceAnnotationPLSFqdns,
+		consts.ServiceAnnotationPLSProxyProtocol,
+		consts.ServiceAnnotationPLSVisibility,
+		consts.ServiceAnnotationPLSAutoApproval}
+	for _, k := range tagKeyList {
+		if _, found := service.Annotations[k]; found {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/provider/azure_utils.go b/pkg/provider/azure_utils.go
index 24090f749..5376e36e0 100644
--- a/pkg/provider/azure_utils.go
+++ b/pkg/provider/azure_utils.go
@@ -234,3 +234,27 @@ func getExtendedLocationTypeFromString(extendedLocationType string) network.Exte
 	}
 	return network.ExtendedLocationTypesEdgeZone
 }
+
+func getBoolValueFromServiceAnnotations(service *v1.Service, key string) bool {
+	if l, found := service.Annotations[key]; found {
+		return strings.EqualFold(strings.TrimSpace(l), consts.TrueAnnotationValue)
+	}
+	return false
+}
+
+func sameContentInSlices(s1 []string, s2 []string) bool {
+	if len(s1) != len(s2) {
+		return false
+	}
+	map1 := make(map[string]int)
+	for _, s := range s1 {
+		map1[s]++
+	}
+	for _, s := range s2 {
+		if v, ok := map1[s]; !ok || v <= 0 {
+			return false
+		}
+		map1[s]--
+	}
+	return true
+}
diff --git a/pkg/provider/azure_wrap.go b/pkg/provider/azure_wrap.go
index ad73f398d..31b983b68 100644
--- a/pkg/provider/azure_wrap.go
+++ b/pkg/provider/azure_wrap.go
@@ -42,6 +42,7 @@ var (
 	nsgCacheTTLDefaultInSeconds          = 120
 	routeTableCacheTTLDefaultInSeconds   = 120
 	publicIPCacheTTLDefaultInSeconds     = 120
+	plsCacheTTLDefaultInSeconds          = 120
 
 	azureNodeProviderIDRE    = regexp.MustCompile(`^azure:///subscriptions/(?:.*)/resourceGroups/(?:.*)/providers/Microsoft.Compute/(?:.*)`)
 	azureResourceGroupNameRE = regexp.MustCompile(`.*/subscriptions/(?:.*)/resourceGroups/(.+)/providers/(?:.*)`)
@@ -175,6 +176,14 @@ func (az *Cloud) getSecurityGroup(crt azcache.AzureCacheReadType) (network.Secur
 	return *(securityGroup.(*network.SecurityGroup)), nil
 }
 
+func (az *Cloud) getPrivateLinkService(frontendIPConfigID *string, crt azcache.AzureCacheReadType) (pls network.PrivateLinkService, err error) {
+	cachedPLS, err := az.plsCache.Get(*frontendIPConfigID, crt)
+	if err != nil {
+		return pls, err
+	}
+	return *(cachedPLS.(*network.PrivateLinkService)), nil
+}
+
 func (az *Cloud) newVMCache() (*azcache.TimedCache, error) {
 	getter := func(key string) (interface{}, error) {
 		// Currently InstanceView request are used by azure_zones, while the calls come after non-InstanceView
@@ -321,6 +330,47 @@ func (az *Cloud) newPIPCache() (*azcache.TimedCache, error) {
 	return azcache.NewTimedcache(time.Duration(az.PublicIPCacheTTLInSeconds)*time.Second, getter)
 }
 
+func (az *Cloud) newPLSCache() (*azcache.TimedCache, error) {
+	// for PLS cache, key is LBFrontendIPConfiguration ID
+	getter := func(key string) (interface{}, error) {
+		ctx, cancel := getContextWithCancel()
+		defer cancel()
+		plsList, err := az.PrivateLinkServiceClient.List(ctx, az.PrivateLinkServiceResourceGroup)
+		exists, rerr := checkResourceExistsFromError(err)
+		if rerr != nil {
+			return nil, rerr.Error()
+		}
+
+		if exists {
+			for i := range plsList {
+				pls := plsList[i]
+				if pls.PrivateLinkServiceProperties == nil {
+					continue
+				}
+				fipConfigs := pls.PrivateLinkServiceProperties.LoadBalancerFrontendIPConfigurations
+				if fipConfigs == nil {
+					continue
+				}
+				for _, fipConfig := range *fipConfigs {
+					if strings.EqualFold(*fipConfig.ID, key) {
+						return &pls, nil
+					}
+				}
+
+			}
+		}
+
+		klog.V(2).Infof("No privateLinkService found for frontendIPConfig %q", key)
+		plsNotExistID := consts.PrivateLinkServiceNotExistID
+		return &network.PrivateLinkService{ID: &plsNotExistID}, nil
+	}
+
+	if az.PlsCacheTTLInSeconds == 0 {
+		az.PlsCacheTTLInSeconds = plsCacheTTLDefaultInSeconds
+	}
+	return azcache.NewTimedcache(time.Duration(az.PlsCacheTTLInSeconds)*time.Second, getter)
+}
+
 func (az *Cloud) useStandardLoadBalancer() bool {
 	return strings.EqualFold(az.LoadBalancerSku, consts.LoadBalancerSkuStandard)
 }


Test patch:
diff --git a/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient_test.go b/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient_test.go
index a5b5aabc6..b781cf024 100644
--- a/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient_test.go
+++ b/pkg/azureclients/privatelinkserviceclient/azure_privatelinkserviceclient_test.go
@@ -311,14 +311,52 @@ func TestDelete(t *testing.T) {
 	}
 }
 
+func TestDeletePEConnection(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	tests := []struct {
+		description  string
+		armClientErr *retry.Error
+		expectedErr  *retry.Error
+	}{
+		{
+			description:  "Delete should report the throttling error",
+			armClientErr: &retry.Error{HTTPStatusCode: http.StatusTooManyRequests},
+			expectedErr:  &retry.Error{HTTPStatusCode: http.StatusTooManyRequests},
+		},
+		{
+			description: "Delete should not report any error if there's no error from arm client",
+		},
+	}
+
+	peConn := getTestPrivateEndpointConnection("pls1", "peconn")
+
+	for _, test := range tests {
+		armClient := mockarmclient.NewMockInterface(ctrl)
+		armClient.EXPECT().DeleteResource(gomock.Any(), to.String(peConn.ID), "").Return(test.armClientErr)
+
+		plsClient := getTestPrivateLinkServiceClient(armClient)
+		rerr := plsClient.DeletePEConnection(context.TODO(), "rg", "pls1", "peconn")
+		assert.Equal(t, test.expectedErr, rerr)
+	}
+}
+
 func getTestPrivateLinkService(name string) network.PrivateLinkService {
 	return network.PrivateLinkService{
-		ID:       to.StringPtr(fmt.Sprintf("/subscriptions/subscriptionID/resourceGroups/rg/providers/"+PLSResourceType+"/%s", name)),
+		ID:       to.StringPtr(fmt.Sprintf("/subscriptions/subscriptionID/resourceGroups/rg/providers/%s/%s", PLSResourceType, name)),
 		Name:     to.StringPtr(name),
 		Location: to.StringPtr("eastus"),
 	}
 }
 
+func getTestPrivateEndpointConnection(PLSName string, PEConnName string) network.PrivateEndpointConnection {
+	return network.PrivateEndpointConnection{
+		ID:   to.StringPtr(fmt.Sprintf("/subscriptions/subscriptionID/resourceGroups/rg/providers/%s/%s/%s/%s", PLSResourceType, PLSName, PEConnResourceType, PEConnName)),
+		Name: to.StringPtr(PEConnName),
+	}
+}
+
 func getTestPrivateLinkServiceClient(armClient armclient.Interface) *Client {
 	rateLimiterReader, rateLimiterWriter := azclients.NewRateLimiter(&azclients.RateLimitConfig{})
 	return &Client{
diff --git a/pkg/provider/azure_loadbalancer_test.go b/pkg/provider/azure_loadbalancer_test.go
index b0ed80823..6e51db03d 100644
--- a/pkg/provider/azure_loadbalancer_test.go
+++ b/pkg/provider/azure_loadbalancer_test.go
@@ -41,6 +41,7 @@ import (
 	cloudprovider "k8s.io/cloud-provider"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/interfaceclient/mockinterfaceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/loadbalancerclient/mockloadbalancerclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/publicipclient/mockpublicipclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/securitygroupclient/mocksecuritygroupclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/subnetclient/mocksubnetclient"
@@ -494,6 +495,12 @@ func TestEnsureLoadBalancerDeleted(t *testing.T) {
 		if c.wrongRGAtDelete {
 			az.LoadBalancerResourceGroup = "nil"
 		}
+
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		err = az.EnsureLoadBalancerDeleted(context.TODO(), testClusterName, &c.service)
 		expectedLBs = make([]network.LoadBalancer, 0)
 		mockLBsClient := mockloadbalancerclient.NewMockInterface(ctrl)
@@ -1411,6 +1418,7 @@ func TestGetServiceLoadBalancer(t *testing.T) {
 						FrontendIPConfigurations: &[]network.FrontendIPConfiguration{
 							{
 								Name: to.StringPtr("aservice1"),
+								ID:   to.StringPtr("as-internal-aservice1"),
 								FrontendIPConfigurationPropertiesFormat: &network.FrontendIPConfigurationPropertiesFormat{
 									PrivateIPAddress: to.StringPtr("1.2.3.4"),
 								},
@@ -1444,6 +1452,11 @@ func TestGetServiceLoadBalancer(t *testing.T) {
 		mockLBsClient.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		az.LoadBalancerClient = mockLBsClient
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), "rg").Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		for _, existingLB := range test.existingLBs {
 			err := az.LoadBalancerClient.CreateOrUpdate(context.TODO(), "rg", *existingLB.Name, existingLB, "")
 			if err != nil {
@@ -4600,7 +4613,10 @@ func TestRemoveFrontendIPConfigurationFromLoadBalancerDelete(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	t.Run("removeFrontendIPConfigurationFromLoadBalancer should remove the unwanted frontend IP configuration and delete the orphaned LB", func(t *testing.T) {
-		fip := &network.FrontendIPConfiguration{Name: to.StringPtr("testCluster")}
+		fip := &network.FrontendIPConfiguration{
+			Name: to.StringPtr("testCluster"),
+			ID:   to.StringPtr("testCluster-fip"),
+		}
 		service := getTestService("svc1", v1.ProtocolTCP, nil, false, 80)
 		lb := getTestLoadBalancer(to.StringPtr("lb"), to.StringPtr("rg"), to.StringPtr("testCluster"), to.StringPtr("testCluster"), service, "standard")
 		bid := "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/networkInterfaces/k8s-agentpool1-00000000-nic-0/ipConfigurations/ipconfig1"
@@ -4617,6 +4633,9 @@ func TestRemoveFrontendIPConfigurationFromLoadBalancerDelete(t *testing.T) {
 		cloud := GetTestCloud(ctrl)
 		mockLBClient := cloud.LoadBalancerClient.(*mockloadbalancerclient.MockInterface)
 		mockLBClient.EXPECT().Delete(gomock.Any(), "rg", "lb").Return(nil)
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := cloud.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+		mockPLSClient.EXPECT().List(gomock.Any(), "rg").Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
 		existingLBs := []network.LoadBalancer{{Name: to.StringPtr("lb")}}
 		err := cloud.removeFrontendIPConfigurationFromLoadBalancer(&lb, existingLBs, fip, "testCluster", &service)
 		assert.NoError(t, err)
@@ -4627,13 +4646,19 @@ func TestRemoveFrontendIPConfigurationFromLoadBalancerUpdate(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	t.Run("removeFrontendIPConfigurationFromLoadBalancer should remove the unwanted frontend IP configuration and update the LB if there are remaining frontend IP configurations", func(t *testing.T) {
-		fip := &network.FrontendIPConfiguration{Name: to.StringPtr("testCluster")}
+		fip := &network.FrontendIPConfiguration{
+			Name: to.StringPtr("testCluster"),
+			ID:   to.StringPtr("testCluster-fip"),
+		}
 		service := getTestService("svc1", v1.ProtocolTCP, nil, false, 80)
 		lb := getTestLoadBalancer(to.StringPtr("lb"), to.StringPtr("rg"), to.StringPtr("testCluster"), to.StringPtr("testCluster"), service, "standard")
 		*lb.FrontendIPConfigurations = append(*lb.FrontendIPConfigurations, network.FrontendIPConfiguration{Name: to.StringPtr("fip1")})
 		cloud := GetTestCloud(ctrl)
 		mockLBClient := cloud.LoadBalancerClient.(*mockloadbalancerclient.MockInterface)
 		mockLBClient.EXPECT().CreateOrUpdate(gomock.Any(), "rg", "lb", gomock.Any(), gomock.Any()).Return(nil)
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := cloud.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+		mockPLSClient.EXPECT().List(gomock.Any(), "rg").Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
 		err := cloud.removeFrontendIPConfigurationFromLoadBalancer(&lb, []network.LoadBalancer{}, fip, "testCluster", &service)
 		assert.NoError(t, err)
 	})
@@ -4807,7 +4832,7 @@ func TestReconcileZonesForFrontendIPConfigs(t *testing.T) {
 			cloud.ZoneClient = zoneClient
 
 			defaultLBFrontendIPConfigName := cloud.getDefaultFrontendIPConfigName(&tc.service)
-			_, dirty, err := cloud.reconcileFrontendIPConfigs("testCluster", &tc.service, &lb, tc.status, true, defaultLBFrontendIPConfigName)
+			_, _, dirty, err := cloud.reconcileFrontendIPConfigs("testCluster", &tc.service, &lb, tc.status, true, defaultLBFrontendIPConfigName)
 			if tc.expectedErr == nil {
 				assert.NoError(t, err)
 			} else {
diff --git a/pkg/provider/azure_privatelinkservice_test.go b/pkg/provider/azure_privatelinkservice_test.go
new file mode 100644
index 000000000..59948f619
--- /dev/null
+++ b/pkg/provider/azure_privatelinkservice_test.go
@@ -0,0 +1,1803 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provider
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
+	"github.com/Azure/go-autorest/autorest/to"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/subnetclient/mocksubnetclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/consts"
+	"sigs.k8s.io/cloud-provider-azure/pkg/retry"
+)
+
+func TestReconcilePrivateLinkService(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc              string
+		annotations       map[string]string
+		wantPLS           bool
+		expectedSubnetGet bool
+		existingSubnet    *network.Subnet
+		expectedPLSList   bool
+		existingPLSList   []network.PrivateLinkService
+		expectedPLSCreate bool
+		expectedPLS       *network.PrivateLinkService
+		expectedPLSDelete bool
+		expectedError     bool
+	}{
+		{
+			desc:    "reconcilePrivateLinkService should do nothing if service does not create any PLS",
+			wantPLS: true,
+		},
+		{
+			desc: "reconcilePrivateLinkService should return error if service requires PLS but needs external LB",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation: "true",
+			},
+			wantPLS:       true,
+			expectedError: true,
+		},
+		{
+			desc: "reconcilePrivateLinkService should create a new PLS if no existing PLS attached to the LB frontend",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			wantPLS:           true,
+			expectedSubnetGet: true,
+			existingSubnet: &network.Subnet{
+				Name: to.StringPtr("subnet"),
+				ID:   to.StringPtr("subnetID"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled,
+				},
+			},
+			expectedPLSList:   true,
+			existingPLSList:   []network.PrivateLinkService{},
+			expectedPLSCreate: true,
+			expectedPLS:       &network.PrivateLinkService{Name: to.StringPtr("testpls")},
+		},
+		{
+			desc: "reconcilePrivateLinkService should report error if existing PLS attached to LB frontEnd is unmanaged",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			wantPLS:         true,
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+						IPConfigurations: &[]network.PrivateLinkServiceIPConfiguration{
+							{
+								PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+									PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+									Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+									Primary:                   to.BoolPtr(true),
+									PrivateIPAddressVersion:   network.IPVersionIPv4,
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedError: true,
+		},
+		{
+			desc: "reconcilePrivateLinkService should report error if service tries to update pls without being an owner",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			wantPLS:         true,
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+						IPConfigurations: &[]network.PrivateLinkServiceIPConfiguration{
+							{
+								PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+									PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+									Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+									Primary:                   to.BoolPtr(true),
+									PrivateIPAddressVersion:   network.IPVersionIPv4,
+								},
+							},
+						},
+					},
+					Tags: map[string]*string{
+						consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+						consts.OwnerServiceTagKey: to.StringPtr("default/test1"),
+					},
+				},
+			},
+			expectedError: true,
+		},
+		{
+			desc: "reconcilePrivateLinkService should share existing pls to a service using the same LB frontEnd without any changes",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+			},
+			wantPLS:         true,
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+						IPConfigurations: &[]network.PrivateLinkServiceIPConfiguration{
+							{
+								PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+									PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+									PrivateIPAddress:          to.StringPtr("10.2.0.4"),
+									Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+									Primary:                   to.BoolPtr(true),
+									PrivateIPAddressVersion:   network.IPVersionIPv4,
+								},
+							},
+						},
+					},
+					Tags: map[string]*string{
+						consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+						consts.OwnerServiceTagKey: to.StringPtr("default/test1"),
+					},
+				},
+			},
+		},
+		{
+			desc: "reconcilePrivateLinkService should not update an existing PLS if everything is same",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			wantPLS:           true,
+			expectedSubnetGet: true,
+			existingSubnet: &network.Subnet{
+				Name: to.StringPtr("subnet"),
+				ID:   to.StringPtr("subnetID"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled,
+				},
+			},
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+						IPConfigurations: &[]network.PrivateLinkServiceIPConfiguration{
+							{
+								PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+									PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+									Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+									Primary:                   to.BoolPtr(true),
+									PrivateIPAddressVersion:   network.IPVersionIPv4,
+								},
+							},
+						},
+					},
+					Tags: map[string]*string{
+						consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+						consts.OwnerServiceTagKey: to.StringPtr("default/test"),
+					},
+				},
+			},
+		},
+		{
+			desc: "reconcilePrivateLinkService should update an existing PLS if some configuration gets changed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:                      "true",
+				consts.ServiceAnnotationLoadBalancerInternal:             "true",
+				consts.ServiceAnnotationPLSName:                          "testpls",
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "2",
+			},
+			wantPLS:           true,
+			expectedSubnetGet: true,
+			existingSubnet: &network.Subnet{
+				Name: to.StringPtr("subnet"),
+				ID:   to.StringPtr("subnetID"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled,
+				},
+			},
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+						IPConfigurations: &[]network.PrivateLinkServiceIPConfiguration{
+							{
+								PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+									PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+									Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+									Primary:                   to.BoolPtr(true),
+									PrivateIPAddressVersion:   network.IPVersionIPv4,
+								},
+							},
+						},
+					},
+					Tags: map[string]*string{
+						consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+						consts.OwnerServiceTagKey: to.StringPtr("default/test"),
+					},
+				},
+			},
+			expectedPLSCreate: true,
+			expectedPLS:       &network.PrivateLinkService{Name: to.StringPtr("testpls")},
+		},
+		{
+			desc: "reconcilePrivateLinkService should not do anything if no existing PLS attached to the LB frontend when deleting",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID2")}},
+					},
+				},
+			},
+		},
+		{
+			desc: "reconcilePrivateLinkService should delete pls when frontend is deleted",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation:          "true",
+				consts.ServiceAnnotationLoadBalancerInternal: "true",
+				consts.ServiceAnnotationPLSName:              "testpls",
+			},
+			expectedPLSList: true,
+			existingPLSList: []network.PrivateLinkService{
+				{
+					Name: to.StringPtr("testpls"),
+					PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+						LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("fipConfigID")}},
+					},
+				},
+			},
+			expectedPLSDelete: true,
+		},
+	}
+	for i, test := range testCases {
+		az := GetTestCloud(ctrl)
+		service := getTestServiceWithAnnotation("test", test.annotations, 80)
+		fipConfig := &network.FrontendIPConfiguration{ID: to.StringPtr("fipConfigID")}
+		clusterName := testClusterName
+
+		mockSubnetsClient := az.SubnetsClient.(*mocksubnetclient.MockInterface)
+		mockPLSsClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+		if test.expectedSubnetGet {
+			mockSubnetsClient.EXPECT().Get(gomock.Any(), "rg", "vnet", "subnet", "").Return(*test.existingSubnet, nil).MaxTimes(2)
+		}
+		if test.expectedPLSList {
+			mockPLSsClient.EXPECT().List(gomock.Any(), "rg").Return(test.existingPLSList, nil).MaxTimes(1)
+		}
+		if test.expectedPLSCreate {
+			mockPLSsClient.EXPECT().CreateOrUpdate(gomock.Any(), "rg", to.String(test.expectedPLS.Name), gomock.Any(), gomock.Any()).Return(nil).Times(1)
+		}
+		if test.expectedPLSDelete {
+			mockPLSsClient.EXPECT().Delete(gomock.Any(), "rg", "testpls").Return(nil).Times(1)
+		}
+		err := az.reconcilePrivateLinkService(clusterName, &service, fipConfig, test.wantPLS)
+		assert.Equal(t, test.expectedError, err != nil, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestDisablePLSNetworkPolicy(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc                 string
+		subnet               network.Subnet
+		expectedSubnetUpdate bool
+		expectedError        bool
+	}{
+		{
+			desc: "disablePLSNetworkPolicy shall not update subnet if pls-network-policy is disabled",
+			subnet: network.Subnet{
+				Name: to.StringPtr("plsSubnet"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled,
+				},
+			},
+			expectedSubnetUpdate: false,
+		},
+		{
+			desc: "disablePLSNetworkPolicy shall update subnet if pls-network-policy is enabled",
+			subnet: network.Subnet{
+				Name: to.StringPtr("plsSubnet"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesEnabled,
+				},
+			},
+			expectedSubnetUpdate: true,
+		},
+	}
+
+	for i, test := range testCases {
+		az := GetTestCloud(ctrl)
+		service := &v1.Service{}
+		service.Annotations = map[string]string{
+			consts.ServiceAnnotationPLSIpConfigurationSubnet: "plsSubnet",
+		}
+		mockSubnetsClient := az.SubnetsClient.(*mocksubnetclient.MockInterface)
+		mockSubnetsClient.EXPECT().Get(gomock.Any(), "rg", "vnet", "plsSubnet", "").Return(test.subnet, nil).Times(1)
+		if test.expectedSubnetUpdate {
+			mockSubnetsClient.EXPECT().CreateOrUpdate(gomock.Any(), "rg", "vnet", "plsSubnet", network.Subnet{
+				Name: to.StringPtr("plsSubnet"),
+				SubnetPropertiesFormat: &network.SubnetPropertiesFormat{
+					PrivateLinkServiceNetworkPolicies: network.VirtualNetworkPrivateLinkServiceNetworkPoliciesDisabled,
+				},
+			}).Return(nil).Times(1)
+		}
+		err := az.disablePLSNetworkPolicy(service)
+		assert.Equal(t, test.expectedError, err != nil, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestSafeDeletePLS(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testCases := []struct {
+		desc          string
+		pls           *network.PrivateLinkService
+		expectedError bool
+	}{
+		{
+			desc: "safeDeletePLS shall delete all PE connections and pls itself",
+			pls: &network.PrivateLinkService{
+				Name: to.StringPtr("testpls"),
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					LoadBalancerFrontendIPConfigurations: &[]network.FrontendIPConfiguration{{ID: to.StringPtr("FipConfigID")}},
+					PrivateEndpointConnections: &[]network.PrivateEndpointConnection{
+						{Name: to.StringPtr("pe1")},
+						{Name: to.StringPtr("pe2")},
+					},
+				},
+			},
+		},
+	}
+
+	for i, test := range testCases {
+		az := GetTestCloud(ctrl)
+		mockPLSsClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+		mockPLSsClient.EXPECT().DeletePEConnection(gomock.Any(), "rg", "testpls", "pe1").Return(nil).Times(1)
+		mockPLSsClient.EXPECT().DeletePEConnection(gomock.Any(), "rg", "testpls", "pe2").Return(nil).Times(1)
+		mockPLSsClient.EXPECT().Delete(gomock.Any(), "rg", "testpls").Return(nil).Times(1)
+		service := getTestService("test1", v1.ProtocolTCP, nil, false, 80)
+		rerr := az.safeDeletePLS(test.pls, &service)
+		assert.Equal(t, test.expectedError, rerr != nil, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestGetPrivateLinkServiceName(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	az := GetTestCloud(ctrl)
+	tests := []struct {
+		desc         string
+		annotations  map[string]string
+		pls          *network.PrivateLinkService
+		fipConfig    *network.FrontendIPConfiguration
+		expectedName string
+		expectedErr  bool
+	}{
+		{
+			desc: "If pls name does not set, sets it as what service configures",
+			pls:  &network.PrivateLinkService{},
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSName: "testpls",
+			},
+			expectedName: "testpls",
+		},
+		{
+			desc: "If pls name does not set, and service does not configure, sets it as default(pls-fipConfigName)",
+			pls:  &network.PrivateLinkService{},
+			fipConfig: &network.FrontendIPConfiguration{
+				Name: to.StringPtr("fipname"),
+			},
+			expectedName: "pls-fipname",
+		},
+		{
+			desc: "If pls name is not equal to service configuration, error should be reported",
+			pls: &network.PrivateLinkService{
+				Name: to.StringPtr("testpls"),
+			},
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSName: "testpls1",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "If pls name is same as service configuration, simply return it",
+			pls: &network.PrivateLinkService{
+				Name: to.StringPtr("testpls"),
+			},
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSName: "testpls",
+			},
+			expectedName: "testpls",
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actualName, err := az.getPrivateLinkServiceName(test.pls, s, test.fipConfig)
+		if test.expectedErr {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+			assert.Equal(t, test.expectedName, actualName, "TestCase[%d]: %s", i, test.desc)
+		}
+	}
+}
+
+func TestGetExpectedPrivateLinkService(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	t.Run("getExpectedPrivateLinkService correctly sets all configurations", func(t *testing.T) {
+		cloud := GetTestCloud(ctrl)
+		service := &v1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "default",
+				Name:      "service",
+				Annotations: map[string]string{
+					consts.ServiceAnnotationPLSIpConfigurationSubnet:         "subnet",
+					consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "3",
+					consts.ServiceAnnotationPLSIpConfigurationIPAddress:      "10.2.0.4 10.2.0.5",
+					consts.ServiceAnnotationPLSFqdns:                         "fqdns1 fqdns2 fqdns3",
+					consts.ServiceAnnotationPLSProxyProtocol:                 "true",
+					consts.ServiceAnnotationPLSVisibility:                    "*",
+					consts.ServiceAnnotationPLSAutoApproval:                  "sub1 sub2 sub3",
+				},
+			},
+		}
+		plsName := "testPLS"
+		clusterName := testClusterName
+		fipConfig := &network.FrontendIPConfiguration{ID: to.StringPtr("fipConfigID")}
+		pls := &network.PrivateLinkService{PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{}}
+		subnetClient := cloud.SubnetsClient.(*mocksubnetclient.MockInterface)
+		subnetClient.EXPECT().Get(gomock.Any(), "rg", "vnet", "subnet", gomock.Any()).Return(
+			network.Subnet{
+				ID:   to.StringPtr("subnetID"),
+				Name: to.StringPtr("subnet"),
+			}, nil).MaxTimes(1)
+
+		dirtyPLS, err := cloud.getExpectedPrivateLinkService(pls, &plsName, &clusterName, service, fipConfig)
+		assert.NoError(t, err)
+		assert.True(t, dirtyPLS)
+
+		assert.Equal(t, pls.Name, &plsName)
+
+		expectedLBFrontendConfig := &[]network.FrontendIPConfiguration{{ID: fipConfig.ID}}
+		assert.Equal(t, pls.LoadBalancerFrontendIPConfigurations, expectedLBFrontendConfig)
+
+		expectedConfigs := []network.PrivateLinkServiceIPConfiguration{
+			{
+				Name: to.StringPtr("subnet-testPLS-static-10.2.0.4"),
+				PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+					PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+					PrivateIPAddress:          to.StringPtr("10.2.0.4"),
+					Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+					Primary:                   to.BoolPtr(true),
+					PrivateIPAddressVersion:   network.IPVersionIPv4,
+				},
+			},
+			{
+				Name: to.StringPtr("subnet-testPLS-static-10.2.0.5"),
+				PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+					PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+					PrivateIPAddress:          to.StringPtr("10.2.0.5"),
+					Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+					Primary:                   to.BoolPtr(false),
+					PrivateIPAddressVersion:   network.IPVersionIPv4,
+				},
+			},
+			{
+				Name: to.StringPtr("subnet-testPLS-dynamic-0"),
+				PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+					PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+					Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+					Primary:                   to.BoolPtr(false),
+					PrivateIPAddressVersion:   network.IPVersionIPv4,
+				},
+			},
+		}
+
+		testSamePLSIpConfigs(t, *pls.IPConfigurations, expectedConfigs)
+
+		expectedFqdns := []string{"fqdns1", "fqdns2", "fqdns3"}
+		assert.Equal(t, *pls.Fqdns, expectedFqdns)
+
+		assert.True(t, *pls.EnableProxyProtocol)
+
+		expectedVisibility := []string{"*"}
+		assert.Equal(t, *pls.Visibility.Subscriptions, expectedVisibility)
+
+		expectedAutoApproval := []string{"sub1", "sub2", "sub3"}
+		assert.Equal(t, *pls.AutoApproval.Subscriptions, expectedAutoApproval)
+	})
+}
+
+func TestReconcilePLSIpConfigs(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	for i, test := range []struct {
+		desc              string
+		annotations       map[string]string
+		existingIPConfigs *[]network.PrivateLinkServiceIPConfiguration
+		expectedIPConfigs *[]network.PrivateLinkServiceIPConfiguration
+		getSubnetError    *retry.Error
+		expectedChanged   bool
+		expectedErr       bool
+	}{
+		{
+			desc: "reconcilePLSIpConfigs should report error when subnet specified by service does not exist",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationSubnet: "subnet",
+			},
+			getSubnetError: &retry.Error{HTTPStatusCode: http.StatusNotFound},
+			expectedErr:    true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should report error when ip count specified is fewer than number of static IPs",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationSubnet:         "subnet",
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "1",
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress:      "10.2.0.4 10.2.0.5",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if its ipConfig is nil",
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if its ipConfig count is different",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "2",
+			},
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-1"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(false),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if its subnetID is different",
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID1")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if ip allocation type is changed from dynamic to static",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "2",
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress:      "10.2.0.4",
+			},
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.4"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.4"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(false),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if ip allocation type is changed from static to dynamic",
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.4"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.4"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should change existingPLS if static ip is changed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "10.2.0.5",
+			},
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.4"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.4"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.5"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.5"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedChanged: true,
+		},
+		{
+			desc: "reconcilePLSIpConfigs should not change existingPLS if ip allocation type is dynamic only",
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-dynamic-0"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodDynamic,
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+		},
+		{
+			desc: "reconcilePLSIpConfigs should not change existingPLS if static ip is exactly same",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "10.2.0.5",
+			},
+			existingIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.5"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.5"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+			expectedIPConfigs: &[]network.PrivateLinkServiceIPConfiguration{
+				{
+					Name: to.StringPtr("subnet-testpls-static-10.2.0.5"),
+					PrivateLinkServiceIPConfigurationProperties: &network.PrivateLinkServiceIPConfigurationProperties{
+						PrivateIPAllocationMethod: network.IPAllocationMethodStatic,
+						PrivateIPAddress:          to.StringPtr("10.2.0.5"),
+						Subnet:                    &network.Subnet{ID: to.StringPtr("subnetID")},
+						Primary:                   to.BoolPtr(true),
+						PrivateIPAddressVersion:   network.IPVersionIPv4,
+					},
+				},
+			},
+		},
+	} {
+		cloud := GetTestCloud(ctrl)
+		service := &v1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:   "default",
+				Name:        "service",
+				Annotations: test.annotations,
+			},
+		}
+		pls := &network.PrivateLinkService{
+			Name: to.StringPtr("testpls"),
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				IPConfigurations: test.existingIPConfigs,
+			},
+		}
+		subnetClient := cloud.SubnetsClient.(*mocksubnetclient.MockInterface)
+		subnetClient.EXPECT().Get(gomock.Any(), "rg", "vnet", "subnet", gomock.Any()).Return(
+			network.Subnet{
+				ID:   to.StringPtr("subnetID"),
+				Name: to.StringPtr("subnet"),
+			}, test.getSubnetError).MaxTimes(1)
+
+		changed, err := cloud.reconcilePLSIpConfigs(pls, service)
+		if test.expectedErr {
+			assert.Error(t, err, "TestCase[%d]: %s", i, test.desc)
+		} else {
+			assert.Equal(t, test.expectedChanged, changed, "TestCase[%d]: %s", i, test.desc)
+			testSamePLSIpConfigs(t, *pls.IPConfigurations, *test.expectedIPConfigs)
+		}
+	}
+}
+
+func testSamePLSIpConfigs(t *testing.T, actual []network.PrivateLinkServiceIPConfiguration, expected []network.PrivateLinkServiceIPConfiguration) {
+	actualIPConfigs := make(map[string]network.PrivateLinkServiceIPConfiguration)
+	expectedIPConfigs := make(map[string]network.PrivateLinkServiceIPConfiguration)
+	for _, ipConfig := range actual {
+		actualIPConfigs[to.String(ipConfig.Name)] = ipConfig
+	}
+	for _, ipConfig := range expected {
+		expectedIPConfigs[to.String(ipConfig.Name)] = ipConfig
+	}
+	assert.Equal(t, actualIPConfigs, expectedIPConfigs)
+}
+
+func TestServiceRequiresPLS(t *testing.T) {
+	tests := []struct {
+		desc        string
+		annotations map[string]string
+		expected    bool
+	}{
+		{
+			desc: "Service with nil annotations should return false",
+		},
+		{
+			desc:        "Service with empty annotations should return false",
+			annotations: map[string]string{},
+		},
+		{
+			desc: "Service with false pls creation annotation should return false",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation: "False",
+			},
+		},
+		{
+			desc: "Service with true pls creation annotation should return true",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSCreation: "True",
+			},
+			expected: true,
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actual := serviceRequiresPLS(s)
+		assert.Equal(t, test.expected, actual, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestReconcilePLSEnableProxyProtocol(t *testing.T) {
+	tests := []struct {
+		desc            string
+		annotations     map[string]string
+		pls             *network.PrivateLinkService
+		expectedChanged bool
+		expectedEnabled *bool
+	}{
+		{
+			desc:        "empty service enableProxyProto and empty pls enableProxyProto should not trigger any change",
+			annotations: map[string]string{},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+			},
+		},
+		{
+			desc: "false service enableProxyProto and empty pls enableProxyProto should not trigger any change",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSProxyProtocol: "False",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+			},
+		},
+		{
+			desc: "false service enableProxyProto and false pls enableProxyProto should not trigger any change",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSProxyProtocol: "False",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					EnableProxyProtocol: to.BoolPtr(false),
+				},
+			},
+			expectedEnabled: to.BoolPtr(false),
+		},
+		{
+			desc: "true service enableProxyProto and true pls enableProxyProto should not trigger any change",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSProxyProtocol: "True",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					EnableProxyProtocol: to.BoolPtr(true),
+				},
+			},
+			expectedEnabled: to.BoolPtr(true),
+		},
+		{
+			desc: "true service enableProxyProto and empty pls enableProxyProto should trigger update",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSProxyProtocol: "True",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+			},
+			expectedEnabled: to.BoolPtr(true),
+			expectedChanged: true,
+		},
+		{
+			desc: "false service enableProxyProto and true pls enableProxyProto should trigger update",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSProxyProtocol: "False",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					EnableProxyProtocol: to.BoolPtr(true),
+				},
+			},
+			expectedEnabled: to.BoolPtr(false),
+			expectedChanged: true,
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		changed := reconcilePLSEnableProxyProtocol(test.pls, s)
+		assert.Equal(t, test.expectedChanged, changed, "TestCase[%d]: %s", i, test.desc)
+		assert.Equal(t, test.expectedEnabled, test.pls.EnableProxyProtocol, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestReconcilePLSFqdn(t *testing.T) {
+	tests := []struct {
+		desc            string
+		annotations     map[string]string
+		pls             *network.PrivateLinkService
+		expectedChanged bool
+		expectedfqdns   *[]string
+	}{
+		{
+			desc:        "Empty service fqdns + empty pls fqdns should not trigger any change",
+			annotations: map[string]string{},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+			},
+		},
+		{
+			desc: "Same service fqdns and pls fqdns should not trigger any change",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "fqdns1 fqdns2 fqdns3",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					Fqdns: &[]string{"fqdns2", "fqdns3", "fqdns1"},
+				},
+			},
+			expectedfqdns: &[]string{"fqdns2", "fqdns3", "fqdns1"},
+		},
+		{
+			desc: "fqdns should be changed according to service - 0",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					Fqdns: &[]string{"fqdns2", "fqdns3", "fqdns1"},
+				},
+			},
+			expectedChanged: true,
+			expectedfqdns:   &[]string{},
+		},
+		{
+			desc: "fqdns should be changed according to service - 1",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "fqdns1 fqdns2",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+			},
+			expectedChanged: true,
+			expectedfqdns:   &[]string{"fqdns1", "fqdns2"},
+		},
+		{
+			desc: "fqdns should be changed according to service - 2",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "fqdns1 fqdns2",
+			},
+			pls: &network.PrivateLinkService{
+				PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+					Fqdns: &[]string{"fqdns2", "fqdns3", "fqdns1"},
+				},
+			},
+			expectedChanged: true,
+			expectedfqdns:   &[]string{"fqdns1", "fqdns2"},
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		changed := reconcilePLSFqdn(test.pls, s)
+		assert.Equal(t, test.expectedChanged, changed, "TestCase[%d]: %s", i, test.desc)
+		assert.Equal(t, test.expectedfqdns, test.pls.Fqdns, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestReconcilePLSVisibility(t *testing.T) {
+	service := v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Annotations: map[string]string{},
+		},
+	}
+	pls := network.PrivateLinkService{
+		PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{},
+	}
+
+	t.Run("reconcilePLSVisibility should change nothing when both visibility and auto-approval are nil", func(t *testing.T) {
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.False(t, changed)
+		assert.NoError(t, err)
+	})
+
+	t.Run("reconcilePLSVisibility should return error if visibility is not * but auto-approval is not empty", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility:   "sub1 sub2",
+			consts.ServiceAnnotationPLSAutoApproval: "sub1",
+		}
+		service.Annotations = annotations
+		_, err := reconcilePLSVisibility(&pls, &service)
+		assert.Error(t, err)
+	})
+
+	t.Run("reconcilePLSVisibility should return not changed if both Visibility and autoApproval are same", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility: "sub1 sub2",
+		}
+		service.Annotations = annotations
+		pls.Visibility = &network.PrivateLinkServicePropertiesVisibility{
+			Subscriptions: &[]string{"sub2", "sub1"},
+		}
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.False(t, changed)
+	})
+
+	t.Run("reconcilePLSVisibility should return not changed if both Visibility and autoApproval are same", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility:   "*",
+			consts.ServiceAnnotationPLSAutoApproval: "sub1 sub2",
+		}
+		service.Annotations = annotations
+		pls.Visibility = &network.PrivateLinkServicePropertiesVisibility{
+			Subscriptions: &[]string{"*"},
+		}
+		pls.AutoApproval = &network.PrivateLinkServicePropertiesAutoApproval{
+			Subscriptions: &[]string{"sub1", "sub2"},
+		}
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.False(t, changed)
+	})
+
+	t.Run("reconcilePLSVisibility should return change pls according to service - 0", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility:   "*",
+			consts.ServiceAnnotationPLSAutoApproval: "sub1 sub2",
+		}
+		pls.Visibility = nil
+		pls.AutoApproval = nil
+		expectedPLS := network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"*"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{"sub1", "sub2"},
+				},
+			},
+		}
+		service.Annotations = annotations
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.True(t, changed)
+		assert.Equal(t, pls, expectedPLS)
+	})
+
+	t.Run("reconcilePLSVisibility should return change pls according to service - 1", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility: "sub1 sub2",
+		}
+		pls.Visibility = nil
+		pls.AutoApproval = nil
+		expectedPLS := network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"sub1", "sub2"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{},
+				},
+			},
+		}
+		service.Annotations = annotations
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.True(t, changed)
+		assert.Equal(t, pls, expectedPLS)
+	})
+
+	t.Run("reconcilePLSVisibility should return change pls according to service - 2", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility: "sub1 sub2",
+		}
+		pls = network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"*"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{"sub1"},
+				},
+			},
+		}
+		expectedPLS := network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"sub1", "sub2"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{},
+				},
+			},
+		}
+		service.Annotations = annotations
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.True(t, changed)
+		assert.Equal(t, pls, expectedPLS)
+	})
+
+	t.Run("reconcilePLSVisibility should return change pls according to service - 3", func(t *testing.T) {
+		annotations := map[string]string{
+			consts.ServiceAnnotationPLSVisibility:   "*",
+			consts.ServiceAnnotationPLSAutoApproval: "sub1 sub2",
+		}
+		pls = network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"*"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{"sub3"},
+				},
+			},
+		}
+		expectedPLS := network.PrivateLinkService{
+			PrivateLinkServiceProperties: &network.PrivateLinkServiceProperties{
+				Visibility: &network.PrivateLinkServicePropertiesVisibility{
+					Subscriptions: &[]string{"*"},
+				},
+				AutoApproval: &network.PrivateLinkServicePropertiesAutoApproval{
+					Subscriptions: &[]string{"sub1", "sub2"},
+				},
+			},
+		}
+		service.Annotations = annotations
+		changed, err := reconcilePLSVisibility(&pls, &service)
+		assert.NoError(t, err)
+		assert.True(t, changed)
+		assert.Equal(t, pls, expectedPLS)
+	})
+}
+
+func TestReconcilePLSTags(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	cloud := GetTestCloud(ctrl)
+	cloud.Tags = "a=x,y=z"
+
+	service := v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "test-ns",
+			Name:      "test-svc",
+		},
+	}
+	pls := network.PrivateLinkService{
+		Tags: map[string]*string{
+			"foo": to.StringPtr("bar"),
+			"a":   to.StringPtr("j"),
+			"m":   to.StringPtr("n"),
+		},
+	}
+	clusterName := testClusterName
+
+	t.Run("reconcilePLSTags should ensure the pls is tagged as configured", func(t *testing.T) {
+		expectedPLS := network.PrivateLinkService{
+			Tags: map[string]*string{
+				consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+				consts.OwnerServiceTagKey: to.StringPtr("test-ns/test-svc"),
+				"foo":                     to.StringPtr("bar"),
+				"a":                       to.StringPtr("x"),
+				"y":                       to.StringPtr("z"),
+				"m":                       to.StringPtr("n"),
+			},
+		}
+		changed := cloud.reconcilePLSTags(&pls, &clusterName, &service)
+		assert.True(t, changed)
+		assert.Equal(t, expectedPLS, pls)
+	})
+
+	t.Run("reconcilePLSTags should delete the old tags if the SystemTags is set", func(t *testing.T) {
+		cloud.SystemTags = "a,foo,b"
+		expectedPLS := network.PrivateLinkService{
+			Tags: map[string]*string{
+				consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+				consts.OwnerServiceTagKey: to.StringPtr("test-ns/test-svc"),
+				"foo":                     to.StringPtr("bar"),
+				"a":                       to.StringPtr("x"),
+				"y":                       to.StringPtr("z"),
+			},
+		}
+		changed := cloud.reconcilePLSTags(&pls, &clusterName, &service)
+		assert.True(t, changed)
+		assert.Equal(t, expectedPLS, pls)
+	})
+
+	t.Run("reconcilePLSTags should support TagsMap", func(t *testing.T) {
+		cloud.SystemTags = "a,foo,b"
+		cloud.TagsMap = map[string]string{"a": "c", "a=b": "c=d", "Y": "zz"}
+		expectedPLS := network.PrivateLinkService{
+			Tags: map[string]*string{
+				consts.ClusterNameTagKey:  to.StringPtr(testClusterName),
+				consts.OwnerServiceTagKey: to.StringPtr("test-ns/test-svc"),
+				"foo":                     to.StringPtr("bar"),
+				"a":                       to.StringPtr("c"),
+				"a=b":                     to.StringPtr("c=d"),
+			},
+		}
+		changed := cloud.reconcilePLSTags(&pls, &clusterName, &service)
+		assert.True(t, changed)
+		assert.Equal(t, expectedPLS, pls)
+	})
+
+	pls.Tags[consts.ClusterNameTagKey] = to.StringPtr("testCluster1")
+	pls.Tags[consts.OwnerServiceTagKey] = to.StringPtr("default/svc")
+
+	t.Run("reconcilePLSTags should respect cluster and owner service tag keys", func(t *testing.T) {
+		expectedPLS := network.PrivateLinkService{
+			Tags: map[string]*string{
+				consts.ClusterNameTagKey:  to.StringPtr("testCluster1"),
+				consts.OwnerServiceTagKey: to.StringPtr("default/svc"),
+				"foo":                     to.StringPtr("bar"),
+				"a":                       to.StringPtr("c"),
+				"a=b":                     to.StringPtr("c=d"),
+				"Y":                       to.StringPtr("zz"),
+			},
+		}
+		changed := cloud.reconcilePLSTags(&pls, &clusterName, &service)
+		assert.True(t, changed)
+		assert.Equal(t, expectedPLS, pls)
+	})
+}
+
+func TestGetPLSSubnetName(t *testing.T) {
+	tests := []struct {
+		desc           string
+		annotations    map[string]string
+		expectedSubnet *string
+	}{
+		{
+			desc: "Service with nil annotations should return nil",
+		},
+		{
+			desc: "Service with empty annotations should return nil",
+		},
+		{
+			desc: "Service with private link subnet specified should return it",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationSubnet: "pls-subnet",
+			},
+			expectedSubnet: to.StringPtr("pls-subnet"),
+		},
+		{
+			desc: "Service with empty private link subnet specified but LB subnet specified should return LB subnet",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationSubnet:   "",
+				consts.ServiceAnnotationLoadBalancerInternalSubnet: "lb-subnet",
+			},
+			expectedSubnet: to.StringPtr("lb-subnet"),
+		},
+		{
+			desc: "Service with LB subnet specified should return it",
+			annotations: map[string]string{
+				consts.ServiceAnnotationLoadBalancerInternalSubnet: "lb-subnet",
+			},
+			expectedSubnet: to.StringPtr("lb-subnet"),
+		},
+		{
+			desc: "Service with both empty subnets specified should return nil",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationSubnet:   "",
+				consts.ServiceAnnotationLoadBalancerInternalSubnet: "",
+			},
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actualSubnet := getPLSSubnetName(s)
+		assert.Equal(t, test.expectedSubnet, actualSubnet, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestGetPLSIPConfigCount(t *testing.T) {
+	tests := []struct {
+		desc            string
+		annotations     map[string]string
+		expectedIPCount int32
+		expectedErr     bool
+	}{
+		{
+			desc:            "Service with nil annotations should return default(1) without any error",
+			expectedIPCount: 1,
+		},
+		{
+			desc:            "Service with empty annotations should return default(1) without any error",
+			annotations:     map[string]string{},
+			expectedIPCount: 1,
+		},
+		{
+			desc: "Service with valid ip count specified should return it",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "6",
+			},
+			expectedIPCount: 6,
+		},
+		{
+			desc: "Service with < 1 ip count specified should return error",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "0",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "Service with > 8 ip count specified should return error",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "9",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "Redundant spaces should be removed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "   4   ",
+			},
+			expectedIPCount: 4,
+		},
+		{
+			desc: "Service with not valid digit string specified should return error",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "test2",
+			},
+			expectedErr: true,
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actualCount, err := getPLSIPConfigCount(s)
+		if test.expectedErr {
+			assert.Error(t, err, "TestCase[%d]: %s", i, test.desc)
+		} else {
+			assert.Equal(t, test.expectedIPCount, actualCount, "TestCase[%d]: %s", i, test.desc)
+			assert.NoError(t, err)
+		}
+	}
+}
+
+func TestGetPLSFqdns(t *testing.T) {
+	tests := []struct {
+		desc          string
+		annotations   map[string]string
+		expectedFqdns []string
+	}{
+		{
+			desc:          "Service with nil annotations should return all empty result without any error",
+			expectedFqdns: []string{},
+		},
+		{
+			desc:          "Service with empty annotations should return all empty result without any error",
+			annotations:   map[string]string{},
+			expectedFqdns: []string{},
+		},
+		{
+			desc: "Service with just 1 fqdns should include it in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "test-fqdns",
+			},
+			expectedFqdns: []string{"test-fqdns"},
+		},
+		{
+			desc: "Service with multiple fqdns should include them in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "test-fqdns1 test-fqdns2 test-fqdns3",
+			},
+			expectedFqdns: []string{"test-fqdns1", "test-fqdns2", "test-fqdns3"},
+		},
+		{
+			desc: "Redundant spaces should be removed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSFqdns: "    test-fqdns1     test-fqdns2    test-fqdns3   ",
+			},
+			expectedFqdns: []string{"test-fqdns1", "test-fqdns2", "test-fqdns3"},
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actual := getPLSFqdns(s)
+		assert.Equal(t, test.expectedFqdns, actual, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestGetPLSVisibility(t *testing.T) {
+	tests := []struct {
+		desc             string
+		annotations      map[string]string
+		expectedVis      []string
+		expectedAllowAll bool
+	}{
+		{
+			desc:        "Service with nil annotations should return all empty result without any error",
+			expectedVis: []string{},
+		},
+		{
+			desc:        "Service with empty annotations should return all empty result without any error",
+			annotations: map[string]string{},
+			expectedVis: []string{},
+		},
+		{
+			desc: "Service with just 1 visibility should include it in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSVisibility: "test-sub-id",
+			},
+			expectedVis: []string{"test-sub-id"},
+		},
+		{
+			desc: "Service with multiple visibilities should include them in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSVisibility: "test-sub-id1 test-sub-id2 test-sub-id3",
+			},
+			expectedVis: []string{"test-sub-id1", "test-sub-id2", "test-sub-id3"},
+		},
+		{
+			desc: "All redundant spaces should be removed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSVisibility: "   test-sub-id1    test-sub-id2   test-sub-id3   ",
+			},
+			expectedVis: []string{"test-sub-id1", "test-sub-id2", "test-sub-id3"},
+		},
+		{
+			desc: "Visibility with * should include it in result, and set allow-all as true",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSVisibility: "*",
+			},
+			expectedVis:      []string{"*"},
+			expectedAllowAll: true,
+		},
+		{
+			desc: "Visibility with * and other sub ids should only include * in result, and set allow-all as true",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSVisibility: "   test-sub-id1     *  test-sub-id2",
+			},
+			expectedVis:      []string{"*"},
+			expectedAllowAll: true,
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actualVis, actualAllowAll := getPLSVisibility(s)
+		assert.Equal(t, test.expectedVis, actualVis, "TestCase[%d]: %s", i, test.desc)
+		assert.Equal(t, test.expectedAllowAll, actualAllowAll, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestGetPLSAutoApproval(t *testing.T) {
+	tests := []struct {
+		desc              string
+		annotations       map[string]string
+		expectedApprovals []string
+	}{
+		{
+			desc:              "Service with nil annotations should return all empty result without any error",
+			expectedApprovals: []string{},
+		},
+		{
+			desc:              "Service with empty annotations should return all empty result without any error",
+			annotations:       map[string]string{},
+			expectedApprovals: []string{},
+		},
+		{
+			desc: "Service with just 1 auto approval should include it in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSAutoApproval: "test-sub-id",
+			},
+			expectedApprovals: []string{"test-sub-id"},
+		},
+		{
+			desc: "Service with multiple auto approvals should include them in result",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSAutoApproval: "test-sub-id1 test-sub-id2 test-sub-id3",
+			},
+			expectedApprovals: []string{"test-sub-id1", "test-sub-id2", "test-sub-id3"},
+		},
+		{
+			desc: "All redundant spaces should be removed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSAutoApproval: "   test-sub-id1    test-sub-id2   test-sub-id3   ",
+			},
+			expectedApprovals: []string{"test-sub-id1", "test-sub-id2", "test-sub-id3"},
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actual := getPLSAutoApproval(s)
+		assert.Equal(t, test.expectedApprovals, actual, "TestCase[%d]: %s", i, test.desc)
+	}
+}
+
+func TestGetPLSStaticIPs(t *testing.T) {
+	tests := []struct {
+		desc              string
+		annotations       map[string]string
+		expectedIPs       map[string]bool
+		expectedPrimaryIP string
+		expectedErr       bool
+	}{
+		{
+			desc:        "Service with nil annotations should return all empty result without any error",
+			expectedIPs: map[string]bool{},
+		},
+		{
+			desc:        "Service with empty annotations should return all empty result without any error",
+			annotations: map[string]string{},
+			expectedIPs: map[string]bool{},
+		},
+		{
+			desc: "Service with just 1 static ip should include it in map and set as primary ip",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "10.2.0.4",
+			},
+			expectedIPs: map[string]bool{
+				"10.2.0.4": true,
+			},
+			expectedPrimaryIP: "10.2.0.4",
+		},
+		{
+			desc: "Service with multiple static ips should include them in map and set the first one as primary ip",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "10.2.0.4 10.2.0.5 10.2.0.6",
+			},
+			expectedIPs: map[string]bool{
+				"10.2.0.4": true,
+				"10.2.0.5": true,
+				"10.2.0.6": true,
+			},
+			expectedPrimaryIP: "10.2.0.4",
+		},
+		{
+			desc: "Service with invalid ip should return error",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "10.2.0.4 300.1.2.9 10.2.0.6",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "Service with ipv6 address should return error",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "fc00:f853:ccd:e793::1",
+			},
+			expectedErr: true,
+		},
+		{
+			desc: "All redundant spaces should be removed",
+			annotations: map[string]string{
+				consts.ServiceAnnotationPLSIpConfigurationIPAddress: "   10.2.0.4    10.2.0.5     10.2.0.6 ",
+			},
+			expectedIPs: map[string]bool{
+				"10.2.0.4": true,
+				"10.2.0.5": true,
+				"10.2.0.6": true,
+			},
+			expectedPrimaryIP: "10.2.0.4",
+		},
+	}
+	for i, test := range tests {
+		s := &v1.Service{}
+		s.Annotations = test.annotations
+		actualIPs, actualPrimaryIP, err := getPLSStaticIPs(s)
+		if test.expectedErr {
+			assert.Error(t, err, "TestCase[%d]: %s", i, test.desc)
+		} else {
+			assert.Equal(t, test.expectedIPs, actualIPs, "TestCase[%d]: %s", i, test.desc)
+			assert.Equal(t, test.expectedPrimaryIP, actualPrimaryIP, "TestCase[%d]: %s", i, test.desc)
+			assert.NoError(t, err)
+		}
+	}
+}
+
+func TestIsManagedPrivateLinkSerivce(t *testing.T) {
+	tests := []struct {
+		desc        string
+		pls         *network.PrivateLinkService
+		clusterName string
+		expected    bool
+	}{
+		{
+			desc: "Private link service with nil tag should return false",
+			pls:  &network.PrivateLinkService{},
+		},
+		{
+			desc: "Private link service with empty tag should return false",
+			pls: &network.PrivateLinkService{
+				Tags: map[string]*string{},
+			},
+		},
+		{
+			desc: "Private link service with different cluster name should return false",
+			pls: &network.PrivateLinkService{
+				Tags: map[string]*string{
+					"k8s-azure-cluster-name": to.StringPtr("test-cluster1"),
+				},
+			},
+			clusterName: "test-cluster",
+		},
+		{
+			desc: "Private link service with same cluster name should return true",
+			pls: &network.PrivateLinkService{
+				Tags: map[string]*string{
+					"k8s-azure-cluster-name": to.StringPtr("test-cluster"),
+				},
+			},
+			clusterName: "test-cluster",
+			expected:    true,
+		},
+	}
+	for i, c := range tests {
+		actual := isManagedPrivateLinkSerivce(c.pls, c.clusterName)
+		assert.Equal(t, actual, c.expected, "TestCase[%d]: %s", i, c.desc)
+	}
+}
+
+func TestGetPrivateLinkServiceOwner(t *testing.T) {
+	tests := []struct {
+		desc     string
+		pls      *network.PrivateLinkService
+		expected string
+	}{
+		{
+			desc: "Private link service with nil tag should return empty string",
+			pls:  &network.PrivateLinkService{},
+		},
+		{
+			desc: "Private link service with empty tag should return empty string",
+			pls: &network.PrivateLinkService{
+				Tags: map[string]*string{},
+			},
+		},
+		{
+			desc: "Private link service with service owner tag should return service owner",
+			pls: &network.PrivateLinkService{
+				Tags: map[string]*string{"k8s-azure-owner-service": to.StringPtr("test-service")},
+			},
+			expected: "test-service",
+		},
+	}
+	for i, c := range tests {
+		actual := getPrivateLinkServiceOwner(c.pls)
+		assert.Equal(t, actual, c.expected, "TestCase[%d]: %s", i, c.desc)
+	}
+}
+
+func TestServiceHasAdditionalConfigs(t *testing.T) {
+	tests := []struct {
+		desc        string
+		annotations map[string]string
+		expected    bool
+	}{
+		{
+			desc:     "Service without any annotations should return false",
+			expected: false,
+		},
+		{
+			desc:        "Service with only pls-create annotation should return false",
+			annotations: map[string]string{consts.ServiceAnnotationPLSCreation: "true"},
+			expected:    false,
+		},
+		{
+			desc:        "Service with pls-name annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSName: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-subnet annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSIpConfigurationSubnet: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-ip-address-count annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSIpConfigurationIPAddressCount: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-ip-address annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSIpConfigurationIPAddress: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-fqdns annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSFqdns: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-proxy-protocol annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSProxyProtocol: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-visibility annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSVisibility: "test"},
+			expected:    true,
+		},
+		{
+			desc:        "Service with pls-auto-approval annotation should return true",
+			annotations: map[string]string{consts.ServiceAnnotationPLSAutoApproval: "test"},
+			expected:    true,
+		},
+	}
+	for i, c := range tests {
+		s := &v1.Service{}
+		s.Annotations = c.annotations
+		actual := serviceHasAdditionalConfigs(s)
+		assert.Equal(t, actual, c.expected, "TestCase[%d]: %s", i, c.desc)
+	}
+}
diff --git a/pkg/provider/azure_ratelimit_test.go b/pkg/provider/azure_ratelimit_test.go
index 85e5e5db0..5881a3518 100644
--- a/pkg/provider/azure_ratelimit_test.go
+++ b/pkg/provider/azure_ratelimit_test.go
@@ -63,6 +63,7 @@ var (
 		"nsgCacheTTLInSeconds": 100,
 		"routeTableCacheTTLInSeconds": 100,
 		"publicIPCacheTTLInSeconds": 100,
+		"plsCacheTTLInSeconds": 100,
 		"location": "location",
 		"maximumLoadBalancerRuleCount": 1,
 		"primaryAvailabilitySetName": "primaryAvailabilitySetName",
@@ -72,6 +73,7 @@ var (
 		"routeTableResourceGroup": "routeTableResourceGroup",
 		"securityGroupName": "securityGroupName",
 		"securityGroupResourceGroup": "securityGroupResourceGroup",
+		"privateLinkServiceResourceGroup": "privateLinkServiceResourceGroup",
 		"subnetName": "subnetName",
 		"subscriptionId": "subscriptionId",
 		"tenantId": "tenantId",
@@ -129,6 +131,7 @@ func TestParseConfig(t *testing.T) {
 		NsgCacheTTLInSeconds:                  100,
 		RouteTableCacheTTLInSeconds:           100,
 		PublicIPCacheTTLInSeconds:             100,
+		PlsCacheTTLInSeconds:                  100,
 		Location:                              "location",
 		MaximumLoadBalancerRuleCount:          1,
 		PrimaryAvailabilitySetName:            "primaryAvailabilitySetName",
@@ -138,6 +141,7 @@ func TestParseConfig(t *testing.T) {
 		RouteTableResourceGroup:               "routeTableResourceGroup",
 		SecurityGroupName:                     "securityGroupName",
 		SecurityGroupResourceGroup:            "securityGroupResourceGroup",
+		PrivateLinkServiceResourceGroup:       "privateLinkServiceResourceGroup",
 		SubnetName:                            "subnetName",
 		UseInstanceMetadata:                   true,
 		VMType:                                "standard",
diff --git a/pkg/provider/azure_test.go b/pkg/provider/azure_test.go
index 577432f81..e141eebc3 100644
--- a/pkg/provider/azure_test.go
+++ b/pkg/provider/azure_test.go
@@ -43,6 +43,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/auth"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/interfaceclient/mockinterfaceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/loadbalancerclient/mockloadbalancerclient"
+	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatelinkserviceclient/mockprivatelinkserviceclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/publicipclient/mockpublicipclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/securitygroupclient/mocksecuritygroupclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/subnetclient/mocksubnetclient"
@@ -312,6 +313,11 @@ func testLoadBalancerServiceDefaultModeSelection(t *testing.T, isInternal bool)
 
 		expectedLBName := setMockLBs(az, ctrl, &expectedLBs, "service", 1, index, isInternal)
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &svc, clusterResources.nodes)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -370,6 +376,11 @@ func testLoadBalancerServiceAutoModeSelection(t *testing.T, isInternal bool) {
 
 		setMockLBs(az, ctrl, &expectedLBs, "service", availabilitySetCount, index, isInternal)
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &svc, clusterResources.nodes)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -442,6 +453,11 @@ func testLoadBalancerServicesSpecifiedSelection(t *testing.T, isInternal bool) {
 
 		setMockLBs(az, ctrl, &expectedLBs, "service", 1, index, isInternal)
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &svc, clusterResources.nodes)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -489,6 +505,11 @@ func testLoadBalancerMaxRulesServices(t *testing.T, isInternal bool) {
 
 		setMockLBs(az, ctrl, &expectedLBs, "service", az.Config.MaximumLoadBalancerRuleCount, index, isInternal)
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &svc, clusterResources.nodes)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -571,6 +592,11 @@ func testLoadBalancerServiceAutoModeDeleteSelection(t *testing.T, isInternal boo
 
 		setMockLBs(az, ctrl, &expectedLBs, "service", availabilitySetCount, index, isInternal)
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		lbStatus, err := az.EnsureLoadBalancer(context.TODO(), testClusterName, &svc, clusterResources.nodes)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -611,6 +637,11 @@ func testLoadBalancerServiceAutoModeDeleteSelection(t *testing.T, isInternal boo
 			t.Errorf("Unexpected number of LB's: Expected (%d) Found (%d)", expectedNumOfLB, lbCount)
 		}
 
+		expectedPLS := make([]network.PrivateLinkService, 0)
+		mockPLSClient := mockprivatelinkserviceclient.NewMockInterface(ctrl)
+		mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+		az.PrivateLinkServiceClient = mockPLSClient
+
 		err := az.EnsureLoadBalancerDeleted(context.TODO(), testClusterName, &svc)
 		if err != nil {
 			t.Errorf("Unexpected error: %q", err)
@@ -755,6 +786,10 @@ func TestReconcileLoadBalancerEditServiceSubnet(t *testing.T) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 	setMockLBs(az, ctrl, &expectedLBs, "service", 1, 1, true)
 
+	expectedPLS := make([]network.PrivateLinkService, 0)
+	mockPLSClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
 	if err != nil {
 		t.Errorf("Unexpected error reconciling initial svc: %q", err)
@@ -935,6 +970,10 @@ func TestReconcileLoadBalancerMultipleServices(t *testing.T) {
 	expectedLBs := make([]network.LoadBalancer, 0)
 	setMockLBs(az, ctrl, &expectedLBs, "service", 1, 1, false)
 
+	expectedPLS := make([]network.PrivateLinkService, 0)
+	mockPLSClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+
 	_, err := az.reconcileLoadBalancer(testClusterName, &svc1, clusterResources.nodes, true /* wantLb */)
 	if err != nil {
 		t.Errorf("Unexpected error: %q", err)
@@ -1043,6 +1082,10 @@ func TestServiceRespectsNoSessionAffinity(t *testing.T) {
 	mockPIPsClient.EXPECT().CreateOrUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockPIPsClient.EXPECT().Get(gomock.Any(), az.ResourceGroup, gomock.Any(), gomock.Any()).Return(expectedPIP, nil).AnyTimes()
 
+	expectedPLS := make([]network.PrivateLinkService, 0)
+	mockPLSClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
 	if err != nil {
 		t.Errorf("Unexpected error reconciling svc1: %q", err)
@@ -1095,6 +1138,10 @@ func TestServiceRespectsClientIPSessionAffinity(t *testing.T) {
 	mockPIPsClient.EXPECT().CreateOrUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 	mockPIPsClient.EXPECT().Get(gomock.Any(), az.ResourceGroup, gomock.Any(), gomock.Any()).Return(expectedPIP, nil).AnyTimes()
 
+	expectedPLS := make([]network.PrivateLinkService, 0)
+	mockPLSClient := az.PrivateLinkServiceClient.(*mockprivatelinkserviceclient.MockInterface)
+	mockPLSClient.EXPECT().List(gomock.Any(), az.Config.ResourceGroup).Return(expectedPLS, nil).MinTimes(1).MaxTimes(1)
+
 	lb, err := az.reconcileLoadBalancer(testClusterName, &svc, clusterResources.nodes, true /* wantLb */)
 	if err != nil {
 		t.Errorf("Unexpected error reconciling svc1: %q", err)
@@ -1948,6 +1995,7 @@ func TestNewCloudFromJSON(t *testing.T) {
 		"resourceGroup": "--resource-group--",
 		"routeTableResourceGroup": "--route-table-resource-group--",
 		"securityGroupResourceGroup": "--security-group-resource-group--",
+		"privateLinkServiceResourceGroup": "--private-link-service-resource-group--",
 		"location": "--location--",
 		"subnetName": "--subnet-name--",
 		"securityGroupName": "--security-group-name--",
@@ -1966,6 +2014,7 @@ func TestNewCloudFromJSON(t *testing.T) {
 		"nsgCacheTTLInSeconds": 100,
 		"routeTableCacheTTLInSeconds": 100,
 		"publicIPCacheTTLInSeconds": 100,
+		"plsCacheTTLInSeconds": 100,
 		"vmType": "vmss",
 		"disableAvailabilitySetNodes": true
 	}`
@@ -2004,6 +2053,7 @@ aadClientCertPassword: --aad-client-cert-password--
 resourceGroup: --resource-group--
 routeTableResourceGroup: --route-table-resource-group--
 securityGroupResourceGroup: --security-group-resource-group--
+privateLinkServiceResourceGroup: --private-link-service-resource-group--
 location: --location--
 subnetName: --subnet-name--
 securityGroupName: --security-group-name--
@@ -2026,6 +2076,7 @@ loadBalancerCacheTTLInSeconds: 100
 nsgCacheTTLInSeconds: 100
 routeTableCacheTTLInSeconds: 100
 publicIPCacheTTLInSeconds: 100
+plsCacheTTLInSeconds: 100
 vmType: vmss
 disableAvailabilitySetNodes: true
 `
@@ -2062,6 +2113,9 @@ func validateConfig(t *testing.T, config string) { //nolint
 	if azureCloud.SecurityGroupResourceGroup != "--security-group-resource-group--" {
 		t.Errorf("got incorrect value for SecurityGroupResourceGroup")
 	}
+	if azureCloud.PrivateLinkServiceResourceGroup != "--private-link-service-resource-group--" {
+		t.Errorf("got incorrect value for PrivateLinkResourceGroup")
+	}
 	if azureCloud.Location != "--location--" {
 		t.Errorf("got incorrect value for Location")
 	}
@@ -2128,6 +2182,9 @@ func validateConfig(t *testing.T, config string) { //nolint
 	if azureCloud.PublicIPCacheTTLInSeconds != 100 {
 		t.Errorf("got incorrect value for publicIPCacheTTLInSeconds")
 	}
+	if azureCloud.PlsCacheTTLInSeconds != 100 {
+		t.Errorf("got incorrect value for plsCacheTTLInSeconds")
+	}
 	if azureCloud.VMType != consts.VMTypeVMSS {
 		t.Errorf("got incorrect value for vmType")
 	}