Repository: kubernetes-sigs/cloud-provider-azure
Pull request #2427 (2022-09-28 14:50:08)

[release-1.1] Get latest service from serviceLister before UpdateLoadBalancer and s…
…erialize LB reconcile

<!--  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 bug
<!--
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:
Avoid race condition between node-sync-loop routine and service reconcile routine.
Avoid UpdateLoadBalancer() using stale service spec to update Load Balancers.
#### 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 #2406 

#### 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
Avoid race condition between node-sync-loop routine and service reconcile routine.
Avoid UpdateLoadBalancer() using stale service spec to update Load Balancers.
```

#### 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/provider/azure.go b/pkg/provider/azure.go
index 8951f4633..36c17829a 100644
--- a/pkg/provider/azure.go
+++ b/pkg/provider/azure.go
@@ -39,6 +39,7 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/flowcontrol"
@@ -341,6 +342,11 @@ type Cloud struct {
 	plsCache *azcache.TimedCache
 
 	*BlobDiskController
+	// Add service lister to always get latest service
+	serviceLister corelisters.ServiceLister
+	// node-sync-loop routine and service-reconcile routine should not update LoadBalancer at the same time
+	serviceReconcileLock sync.Mutex
+
 	*ManagedDiskController
 	*controllerCommon
 }
@@ -1003,6 +1009,8 @@ func (az *Cloud) SetInformers(informerFactory informers.SharedInformerFactory) {
 		},
 	})
 	az.nodeInformerSynced = nodeInformer.HasSynced
+
+	az.serviceLister = informerFactory.Core().V1().Services().Lister()
 }
 
 // updateNodeCaches updates local cache for node's zones and external resource groups.
diff --git a/pkg/provider/azure_loadbalancer.go b/pkg/provider/azure_loadbalancer.go
index c3e729545..457580ead 100644
--- a/pkg/provider/azure_loadbalancer.go
+++ b/pkg/provider/azure_loadbalancer.go
@@ -31,6 +31,7 @@ import (
 	"github.com/Azure/go-autorest/autorest/to"
 
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	cloudprovider "k8s.io/cloud-provider"
@@ -186,6 +187,11 @@ func (az *Cloud) EnsureLoadBalancer(ctx context.Context, clusterName string, ser
 	// When a client updates the internal load balancer annotation,
 	// the service may be switched from an internal LB to a public one, or vise versa.
 	// Here we'll firstly ensure service do not lie in the opposite LB.
+
+	// Serialize service reconcile process
+	az.serviceReconcileLock.Lock()
+	defer az.serviceReconcileLock.Unlock()
+
 	var err error
 	serviceName := getServiceName(service)
 	mc := metrics.NewMetricContext("services", "ensure_loadbalancer", az.ResourceGroup, az.getNetworkResourceSubscriptionID(), serviceName)
@@ -206,8 +212,25 @@ func (az *Cloud) EnsureLoadBalancer(ctx context.Context, clusterName string, ser
 	return lbStatus, nil
 }
 
+func (az *Cloud) getLatestService(service *v1.Service) (*v1.Service, bool, error) {
+	latestService, err := az.serviceLister.Services(service.Namespace).Get(service.Name)
+	switch {
+	case apierrors.IsNotFound(err):
+		// service absence in store means the service deletion is caught by watcher
+		return nil, false, nil
+	case err != nil:
+		return nil, false, err
+	default:
+		return latestService.DeepCopy(), true, nil
+	}
+}
+
 // UpdateLoadBalancer updates hosts under the specified load balancer.
 func (az *Cloud) UpdateLoadBalancer(ctx context.Context, clusterName string, service *v1.Service, nodes []*v1.Node) error {
+	// Serialize service reconcile process
+	az.serviceReconcileLock.Lock()
+	defer az.serviceReconcileLock.Unlock()
+
 	var err error
 	serviceName := getServiceName(service)
 	mc := metrics.NewMetricContext("services", "update_loadbalancer", az.ResourceGroup, az.getNetworkResourceSubscriptionID(), serviceName)
@@ -218,6 +241,17 @@ func (az *Cloud) UpdateLoadBalancer(ctx context.Context, clusterName string, ser
 		klog.V(5).InfoS("UpdateLoadBalancer Finish", "service", serviceName, "cluster", clusterName, "service_spec", service, "error", err)
 	}()
 
+	// In case UpdateLoadBalancer gets stale service spec, retrieve the latest from lister
+	service, serviceExists, err := az.getLatestService(service)
+	if err != nil {
+		return fmt.Errorf("UpdateLoadBalancer: failed to get latest service %s: %w", service.Name, err)
+	}
+	if !serviceExists {
+		isOperationSucceeded = true
+		klog.V(2).Infof("UpdateLoadBalancer: skipping service %s because service is going to be deleted", service.Name)
+		return nil
+	}
+
 	shouldUpdateLB := az.shouldUpdateLoadBalancer(clusterName, service, nodes)
 	if !shouldUpdateLB {
 		isOperationSucceeded = true
@@ -241,6 +275,10 @@ func (az *Cloud) UpdateLoadBalancer(ctx context.Context, clusterName string, ser
 // have multiple underlying components, meaning a Get could say that the LB
 // doesn't exist even if some part of it is still laying around.
 func (az *Cloud) EnsureLoadBalancerDeleted(ctx context.Context, clusterName string, service *v1.Service) error {
+	// Serialize service reconcile process
+	az.serviceReconcileLock.Lock()
+	defer az.serviceReconcileLock.Unlock()
+
 	var err error
 	isInternal := requiresInternalLoadBalancer(service)
 	serviceName := getServiceName(service)
@@ -2836,7 +2874,7 @@ func (az *Cloud) getExpectedSecurityRules(wantLb bool, ports []v1.ServicePort, s
 
 func (az *Cloud) shouldUpdateLoadBalancer(clusterName string, service *v1.Service, nodes []*v1.Node) bool {
 	_, _, existsLb, _ := az.getServiceLoadBalancer(service, clusterName, nodes, false, []network.LoadBalancer{})
-	return existsLb && service.ObjectMeta.DeletionTimestamp == nil
+	return existsLb && service.ObjectMeta.DeletionTimestamp == nil && service.Spec.Type == v1.ServiceTypeLoadBalancer
 }
 
 func logSafe(s *string) string {


Test patch:
diff --git a/pkg/provider/azure_loadbalancer_test.go b/pkg/provider/azure_loadbalancer_test.go
index 06f4b5876..e66b9e262 100644
--- a/pkg/provider/azure_loadbalancer_test.go
+++ b/pkg/provider/azure_loadbalancer_test.go
@@ -4236,30 +4236,42 @@ func TestShouldUpdateLoadBalancer(t *testing.T) {
 	testCases := []struct {
 		desc                   string
 		lbHasDeletionTimestamp bool
+		serviceType            v1.ServiceType
 		existsLb               bool
 		expectedOutput         bool
 	}{
 		{
 			desc:                   "should update a load balancer that does not have a deletion timestamp and exists in Azure",
 			lbHasDeletionTimestamp: false,
+			serviceType:            v1.ServiceTypeLoadBalancer,
 			existsLb:               true,
 			expectedOutput:         true,
 		},
 		{
 			desc:                   "should not update a load balancer that is being deleted / already deleted in K8s",
 			lbHasDeletionTimestamp: true,
+			serviceType:            v1.ServiceTypeLoadBalancer,
+			existsLb:               true,
+			expectedOutput:         false,
+		},
+		{
+			desc:                   "should not update a load balancer that is no longer LoadBalancer type in K8s",
+			lbHasDeletionTimestamp: false,
+			serviceType:            v1.ServiceTypeClusterIP,
 			existsLb:               true,
 			expectedOutput:         false,
 		},
 		{
 			desc:                   "should not update a load balancer that does not exist in Azure",
 			lbHasDeletionTimestamp: false,
+			serviceType:            v1.ServiceTypeLoadBalancer,
 			existsLb:               false,
 			expectedOutput:         false,
 		},
 		{
 			desc:                   "should not update a load balancer that has a deletion timestamp and does not exist in Azure",
 			lbHasDeletionTimestamp: true,
+			serviceType:            v1.ServiceTypeLoadBalancer,
 			existsLb:               false,
 			expectedOutput:         false,
 		},
@@ -4268,6 +4280,7 @@ func TestShouldUpdateLoadBalancer(t *testing.T) {
 	for i, test := range testCases {
 		az := GetTestCloud(ctrl)
 		service := getTestService("test1", v1.ProtocolTCP, nil, false, 80)
+		service.Spec.Type = test.serviceType
 		setMockPublicIPs(az, ctrl, 1)
 		mockLBsClient := mockloadbalancerclient.NewMockInterface(ctrl)
 		az.LoadBalancerClient = mockLBsClient