Repository: kubernetes-sigs/cloud-provider-azure
Pull request #1691 (2022-05-07 05:26:58)

[Release-1.1]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.1 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 #872 
#### 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

```



Patch:
diff --git a/pkg/auth/azure_auth.go b/pkg/auth/azure_auth.go
index c641f5cd0..730a0a2c5 100644
--- a/pkg/auth/azure_auth.go
+++ b/pkg/auth/azure_auth.go
@@ -100,17 +100,6 @@ func GetServicePrincipalToken(config *AzureAuthConfig, env *azure.Environment, r
 		}
 		if len(config.UserAssignedIdentityID) > 0 {
 			klog.V(4).Info("azure: using User Assigned MSI ID to retrieve access token")
-			resourceID, err := azure.ParseResourceID(config.UserAssignedIdentityID)
-			if err == nil &&
-				strings.EqualFold(resourceID.Provider, "Microsoft.ManagedIdentity") &&
-				strings.EqualFold(resourceID.ResourceType, "userAssignedIdentities") {
-				klog.V(4).Info("azure: User Assigned MSI ID is resource ID")
-				return adal.NewServicePrincipalTokenFromMSIWithIdentityResourceID(msiEndpoint,
-					resource,
-					config.UserAssignedIdentityID)
-			}
-
-			klog.V(4).Info("azure: User Assigned MSI ID is client ID. Resource ID parsing error: %+v", err)
 			return adal.NewServicePrincipalTokenFromMSIWithUserAssignedID(msiEndpoint,
 				resource,
 				config.UserAssignedIdentityID)
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 857c1c18b..e49251ba6 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -54,6 +54,7 @@ import (
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatednsclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privatednszonegroupclient"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azureclients/privateendpointclient"
+	"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"
@@ -226,6 +227,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
@@ -238,6 +241,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 {
@@ -287,6 +292,7 @@ type Cloud struct {
 	privatednsclient                privatednsclient.Interface
 	privatednszonegroupclient       privatednszonegroupclient.Interface
 	virtualNetworkLinksClient       virtualnetworklinksclient.Interface
+	PrivateLinkServiceClient        privatelinkserviceclient.Interface
 
 	ResourceRequestBackoff wait.Backoff
 	metadata               *InstanceMetadataService
@@ -331,6 +337,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
@@ -478,6 +486,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
@@ -638,6 +650,11 @@ func (az *Cloud) initCaches() (err error) {
 		return err
 	}
 
+	az.plsCache, err = az.newPLSCache()
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -796,6 +813,7 @@ func (az *Cloud) configAzureClients(
 	az.privatednsclient = privatednsclient.New(azClientConfig)
 	az.privatednszonegroupclient = privatednszonegroupclient.New(azClientConfig)
 	az.virtualNetworkLinksClient = virtualnetworklinksclient.New(azClientConfig)
+	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 12e53a7ea..86270d06e 100644
--- a/pkg/provider/azure_backoff.go
+++ b/pkg/provider/azure_backoff.go
@@ -520,3 +520,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 4417826df..503fe06af 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"
@@ -67,19 +68,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 },
@@ -101,12 +103,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()
 
 	_ = initDiskControllers(az)
 
diff --git a/pkg/provider/azure_loadbalancer.go b/pkg/provider/azure_loadbalancer.go
index 86a6269c8..e77cfef1c 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)
@@ -1555,7 +1569,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
 	}
@@ -1607,6 +1621,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 {
@@ -1743,13 +1773,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
 	}
@@ -1762,12 +1793,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:
@@ -1785,6 +1816,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
 				}
@@ -1805,10 +1837,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
@@ -1818,7 +1851,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 {
@@ -1833,11 +1866,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{
@@ -1866,12 +1899,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},
@@ -1887,7 +1920,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)
@@ -1900,7 +1933,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 1b5dd1e80..3c2911246 100644
--- a/pkg/provider/azure_utils.go
+++ b/pkg/provider/azure_utils.go
@@ -235,3 +235,27 @@ func getServiceAdditionalPublicIPs(service *v1.Service) ([]string, error) {
 
 	return result, nil
 }
+
+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/auth/azure_auth_test.go b/pkg/auth/azure_auth_test.go
index 435b4956b..89aebdb47 100644
--- a/pkg/auth/azure_auth_test.go
+++ b/pkg/auth/azure_auth_test.go
@@ -65,13 +65,13 @@ func TestGetServicePrincipalTokenFromMSIWithUserAssignedID(t *testing.T) {
 	configs := []*AzureAuthConfig{
 		{
 			UseManagedIdentityExtension: true,
-			UserAssignedIdentityID:      "00000000-0000-0000-0000-000000000000",
+			UserAssignedIdentityID:      "UserAssignedIdentityID",
 		},
 		// The Azure service principal is ignored when
 		// UseManagedIdentityExtension is set to true
 		{
 			UseManagedIdentityExtension: true,
-			UserAssignedIdentityID:      "00000000-0000-0000-0000-000000000000",
+			UserAssignedIdentityID:      "UserAssignedIdentityID",
 			TenantID:                    "TenantID",
 			AADClientID:                 "AADClientID",
 			AADClientSecret:             "AADClientSecret",
@@ -110,55 +110,6 @@ func TestGetServicePrincipalTokenFromMSIWithUserAssignedID(t *testing.T) {
 	}
 }
 
-func TestGetServicePrincipalTokenFromMSIWithIdentityResourceID(t *testing.T) {
-	configs := []*AzureAuthConfig{
-		{
-			UseManagedIdentityExtension: true,
-			UserAssignedIdentityID:      "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/ua",
-		},
-		// The Azure service principal is ignored when
-		// UseManagedIdentityExtension is set to true
-		{
-			UseManagedIdentityExtension: true,
-			UserAssignedIdentityID:      "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.ManagedIdentity/userAssignedIdentities/ua",
-			TenantID:                    "TenantID",
-			AADClientID:                 "AADClientID",
-			AADClientSecret:             "AADClientSecret",
-		},
-	}
-	env := &azure.PublicCloud
-
-	// msiEndpointEnv and msiSecretEnv are required because autorest/adal library requires IMDS endpoint to be available.
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		assert.Equal(t, "GET", r.Method)
-		w.WriteHeader(http.StatusOK)
-		_, err := w.Write([]byte("{}"))
-		assert.NoError(t, err)
-	}))
-	originalEnv := os.Getenv(msiEndpointEnv)
-	originalSecret := os.Getenv(msiSecretEnv)
-	os.Setenv(msiEndpointEnv, server.URL)
-	os.Setenv(msiSecretEnv, "secret")
-	defer func() {
-		server.Close()
-		os.Setenv(msiEndpointEnv, originalEnv)
-		os.Setenv(msiSecretEnv, originalSecret)
-	}()
-
-	for _, config := range configs {
-		token, err := GetServicePrincipalToken(config, env, "")
-		assert.NoError(t, err)
-
-		msiEndpoint, err := adal.GetMSIVMEndpoint()
-		assert.NoError(t, err)
-
-		spt, err := adal.NewServicePrincipalTokenFromMSIWithIdentityResourceID(msiEndpoint,
-			env.ServiceManagementEndpoint, config.UserAssignedIdentityID)
-		assert.NoError(t, err)
-		assert.Equal(t, token, spt)
-	}
-}
-
 func TestGetServicePrincipalTokenFromMSI(t *testing.T) {
 	configs := []*AzureAuthConfig{
 		{
diff --git a/pkg/provider/azure_loadbalancer_test.go b/pkg/provider/azure_loadbalancer_test.go
index c72350e08..d1af3aa12 100644
--- a/pkg/provider/azure_loadbalancer_test.go
+++ b/pkg/provider/azure_loadbalancer_test.go
@@ -42,6 +42,7 @@ import (
 
 	"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"
@@ -495,6 +496,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)
@@ -1412,6 +1419,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"),
 								},
@@ -1445,6 +1453,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 {
@@ -4684,7 +4697,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"
@@ -4701,6 +4717,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)
@@ -4711,13 +4730,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)
 	})
@@ -4891,7 +4916,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 f77aa25ad..3ba565254 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",
@@ -137,6 +139,7 @@ func TestParseConfig(t *testing.T) {
 		NsgCacheTTLInSeconds:                  100,
 		RouteTableCacheTTLInSeconds:           100,
 		PublicIPCacheTTLInSeconds:             100,
+		PlsCacheTTLInSeconds:                  100,
 		Location:                              "location",
 		MaximumLoadBalancerRuleCount:          1,
 		PrimaryAvailabilitySetName:            "primaryAvailabilitySetName",
@@ -146,6 +149,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 028c6411b..d0740fee5 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")
 	}