gce: Implement fine-grained resource-specific locking for L4 ILBs (#1088)

Introduces fine-grained resource-specific locking for internal load balancer (L4 ILB) GCE resources:
- Replaces the single global coarse-grained mutex (sharedResourceLock) with isolated, resource-scoped mutexes for Firewalls, HealthChecks, and InstanceGroups.
- Relocates the coarse-grained global lock acquisition to the top-level entry points of the ILB operations (ensureInternalLoadBalancer, updateInternalLoadBalancer, and ensureInternalLoadBalancerDeleted) when fine-grained locking is disabled.
- Refactors lockResourceIfShared to return a no-op unlock function when the fine-grained resource locking feature gate is disabled, preventing multi-level locking overhead and deadlock risks inside inner helpers.
- Includes comprehensive unit tests to verify proper isolation, lock scoping, and seamless fallback behavior when the feature gate is disabled.

Author: 08volt

cmd/cloud-controller-manager/main.go

@@ -81,6 +81,9 @@ var (
 
 	// enableGKETenantController enables the gke-tenant-controller-manager.
 	enableGKETenantController bool
+
+	// enableL4ILBFineGrainedLocks enables resource-specific locking for L4 ILB.
+	enableL4ILBFineGrainedLocks bool
 )
 
 func main() {
@@ -102,6 +105,7 @@ func main() {
 	cloudProviderFS.BoolVar(&enableL4DenyFirewall, "enable-l4-deny-firewall", false, "Enable creation and updates of Deny VPC Firewall Rules for L4 external load balancers. Requires --enable-pinhole and --enable-l4-deny-firewall-rollback-cleanup to be true.")
 	cloudProviderFS.BoolVar(&enableL4DenyFirewallRollbackCleanup, "enable-l4-deny-firewall-rollback-cleanup", false, "Enable cleanup codepath of the deny firewalls for rollback. The reason for it not being enabled by default is the additional GCE API calls that are made for checking if the deny firewalls exist/deletion which will eat up the quota unnecessarily.")
 	cloudProviderFS.BoolVar(&enableGKETenantController, "enable-gke-tenant-controller", false, "Enables the GKE Tenant Controller Manager for Multi-Tenancy.")
+	cloudProviderFS.BoolVar(&enableL4ILBFineGrainedLocks, "enable-l4-ilb-fine-grained-lock", false, "Enable resource-specific locking for L4 ILB")
 
 	// add new controllers and initializers
 	nodeIpamController := nodeIPAMController{}
@@ -227,5 +231,16 @@ func cloudInitializer(config *cloudcontrollerconfig.CompletedConfig) cloudprovid
 		gceCloud.SetEnableL4DenyFirewallRule(enableL4DenyFirewall, enableL4DenyFirewallRollbackCleanup)
 	}
 
+	if enableL4ILBFineGrainedLocks {
+		gceCloud, ok := (cloud).(*gce.Cloud)
+		if !ok {
+			klog.Fatalf("enable-l4-ilb-fine-grained-lock requires GCE cloud provider, but got %T", cloud)
+		}
+		gceCloud.SetEnableL4ILBFineGrainedLocks(true)
+	}
+
+	// Record feature gate metrics
+	gce.RecordFeatureGateMetrics(enableL4ILBFineGrainedLocks)
+
 	return cloud
 }

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/spf13/pflag v1.0.10
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/oauth2 v0.36.0
+	golang.org/x/sync v0.21.0 // indirect
 	google.golang.org/api v0.284.0
 	gopkg.in/gcfg.v1 v1.2.3
 	gopkg.in/warnings.v0 v0.1.2 // indirect
@@ -131,7 +132,6 @@ require (
 	golang.org/x/crypto v0.51.0 // indirect
 	golang.org/x/mod v0.35.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.21.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/term v0.43.0 // indirect
 	golang.org/x/text v0.37.0 // indirect

go.work.sum

@@ -5,6 +5,7 @@ github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1/go.mod h1:Zanoh4+gvIgluNqcfMVTJueD4wSS5hT7zTt4Mrutd90=
 github.com/onsi/gomega v1.38.2/go.mod h1:W2MJcYxRGV63b418Ai34Ud0hEdTVXq9NW9+Sx6uXf3k=
 go.opentelemetry.io/proto/otlp v1.7.1/go.mod h1:b2rVh6rfI/s2pHWNlB7ILJcRALpcNDzKhACevjI+ZnE=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto=
 k8s.io/pod-security-admission v0.36.0 h1:YgVsB5KFiUtZfHgcLf/GPGGR9KgoXN4/loadBLCRvhY=
 k8s.io/pod-security-admission v0.36.0/go.mod h1:Brj/48uHTUApss1AaehnCw0dgI1Pxk/RAOo1oSNLqhI=

providers/gce/gce.go

@@ -180,11 +180,14 @@ type Cloud struct {
 	// resources are only created in zones with active node capacity.
 	nodeZones          map[string]sets.String
 	nodeInformerSynced cache.InformerSynced
+
 	// sharedResourceLock is used to serialize GCE operations that may mutate shared state to
 	// prevent inconsistencies. For example, load balancers manipulation methods will take the
 	// lock to prevent shared resources from being prematurely deleted while the operation is
 	// in progress.
 	sharedResourceLock sync.Mutex
+	// sharedResourceLocks is a concurrent map used for resource-specific fine-grained locking of shared resources (e.g. InstanceGroups, shared HealthChecks).
+	sharedResourceLocks sync.Map // map[string]*sync.Mutex
 	// AlphaFeatureGate gates gce alpha features in Cloud instance.
 	// Related wrapper functions that interacts with gce alpha api should examine whether
 	// the corresponding api is enabled.
@@ -224,6 +227,67 @@ type Cloud struct {
 
 	// enableL4DenyFirewallRollbackCleanup
 	enableL4DenyFirewallRollbackCleanup bool
+
+	// enableL4ILBFineGrainedLocks enables fine-grained resource-specific locking
+	enableL4ILBFineGrainedLocks bool
+}
+
+type SharedResourceType string
+
+const (
+	ResourceTypeHealthCheck   SharedResourceType = "hc"
+	ResourceTypeInstanceGroup SharedResourceType = "ig"
+	ResourceTypeFirewall      SharedResourceType = "fw"
+)
+
+func (g *Cloud) getLockForResource(resType SharedResourceType, name string) *sync.Mutex {
+	key := string(resType) + ":" + name
+	if v, ok := g.sharedResourceLocks.Load(key); ok {
+		return v.(*sync.Mutex)
+	}
+	v, _ := g.sharedResourceLocks.LoadOrStore(key, &sync.Mutex{})
+	return v.(*sync.Mutex)
+}
+
+// lockSharedResourcesIfCoarse acquires the global sharedResourceLock when fine-grained
+// locking is disabled, preserving the legacy coarse locking behavior.
+// It returns a function to defer for unlocking.
+func (g *Cloud) lockSharedResourcesIfCoarse() func() {
+	if g.enableL4ILBFineGrainedLocks {
+		return func() { /* no-op */ }
+	}
+	g.sharedResourceLock.Lock()
+	return g.sharedResourceLock.Unlock
+}
+
+// lockResourceIfShared is a helper function for acquiring locks on shared resources.
+// If fine-grained locking is disabled or the resource is not shared, it does nothing.
+func (g *Cloud) lockResourceIfShared(shared bool, resType SharedResourceType, name string) func() {
+	if !g.enableL4ILBFineGrainedLocks || !shared {
+		return func() { /* no-op */ }
+	}
+	lock := g.getLockForResource(resType, name)
+	lock.Lock()
+	return lock.Unlock
+}
+
+// lockInstanceGroup locks the shared unmanaged instance group in the specified zone.
+// Since instance groups are always shared across the cluster, this locks unconditionally.
+// It returns a function to defer for unlocking.
+func (g *Cloud) lockInstanceGroup(igName, zone string) func() {
+	return g.lockResourceIfShared(true, ResourceTypeInstanceGroup, igName+"-"+zone)
+}
+
+// lockHealthCheck locks a health check resource by name.
+// It returns a function to defer for unlocking.
+func (g *Cloud) lockHealthCheck(hcName string, shared bool) func() {
+	return g.lockResourceIfShared(shared, ResourceTypeHealthCheck, hcName)
+}
+
+// lockFirewall locks a firewall resource by name.
+// It returns a function to defer for unlocking.
+func (g *Cloud) lockFirewall(fwName string, shared bool) func() {
+	return g.lockResourceIfShared(shared, ResourceTypeFirewall, fwName)
 }
 
 // ConfigGlobal is the in memory representation of the gce.conf config data
@@ -923,6 +987,10 @@ func (g *Cloud) SetEnableL4DenyFirewallRule(firewallEnabled, rollbackEnabled boo
 	g.enableL4DenyFirewallRollbackCleanup = rollbackEnabled
 }
 
+func (g *Cloud) SetEnableL4ILBFineGrainedLocks(enabled bool) {
+	g.enableL4ILBFineGrainedLocks = enabled
+}
+
 // getProjectsBasePath returns the compute API endpoint with the `projects/` element.
 // The suffix must be added when generating compute resource urls.
 func getProjectsBasePath(basePath string) string {

providers/gce/gce_loadbalancer_internal.go

@@ -132,9 +132,8 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
 		}
 	}
 
-	// Lock the sharedResourceLock to prevent any deletions of shared resources while assembling shared resources here
-	g.sharedResourceLock.Lock()
-	defer g.sharedResourceLock.Unlock()
+	unlock := g.lockSharedResourcesIfCoarse()
+	defer unlock()
 
 	// Ensure health check exists before creating the backend service. The health check is shared
 	// if externalTrafficPolicy=Cluster.
@@ -250,7 +249,7 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
 
 	// Delete the previous internal load balancer resources if necessary
 	if existingBackendService != nil {
-		g.clearPreviousInternalResources(svc, loadBalancerName, existingBackendService, backendServiceName, hcName)
+		g.clearPreviousInternalResources(svc, loadBalancerName, clusterID, existingBackendService, backendServiceName, hcName)
 	}
 
 	serviceState.InSuccess = true
@@ -320,7 +319,7 @@ func truncateList[T any](l []T, max int) []T {
 	return l[:max]
 }
 
-func (g *Cloud) clearPreviousInternalResources(svc *v1.Service, loadBalancerName string, existingBackendService *compute.BackendService, expectedBSName, expectedHCName string) {
+func (g *Cloud) clearPreviousInternalResources(svc *v1.Service, loadBalancerName, clusterID string, existingBackendService *compute.BackendService, expectedBSName, expectedHCName string) {
 	// If a new backend service was created, delete the old one.
 	if existingBackendService.Name != expectedBSName {
 		klog.V(2).Infof("clearPreviousInternalResources(%v): expected backend service %q does not match previous %q - deleting backend service", loadBalancerName, expectedBSName, existingBackendService.Name)
@@ -334,7 +333,8 @@ func (g *Cloud) clearPreviousInternalResources(svc *v1.Service, loadBalancerName
 		existingHCName := getNameFromLink(existingBackendService.HealthChecks[0])
 		if existingHCName != expectedHCName {
 			klog.V(2).Infof("clearPreviousInternalResources(%v): expected health check %q does not match previous %q - deleting health check", loadBalancerName, expectedHCName, existingHCName)
-			if err := g.teardownInternalHealthCheckAndFirewall(svc, existingHCName); err != nil {
+			shared := isSharedHealthCheckName(existingHCName, clusterID)
+			if err := g.teardownInternalHealthCheckAndFirewall(svc, existingHCName, shared); err != nil {
 				klog.Warningf("clearPreviousInternalResources: could not delete existing healthcheck: %v, err: %v", existingHCName, err)
 			}
 		}
@@ -362,8 +362,8 @@ func (g *Cloud) updateInternalLoadBalancer(clusterName, clusterID string, svc *v
 	if err := g.processMixedProtocolCheck(context.TODO(), svc, true); err != nil {
 		return err
 	}
-	g.sharedResourceLock.Lock()
-	defer g.sharedResourceLock.Unlock()
+	unlock := g.lockSharedResourcesIfCoarse()
+	defer unlock()
 
 	igName := makeInstanceGroupName(clusterID)
 	igLinks, err := g.ensureInternalInstanceGroups(igName, nodes)
@@ -395,14 +395,15 @@ func (g *Cloud) ensureInternalLoadBalancerDeleted(clusterName, clusterID string,
 	}
 
 	loadBalancerName := g.GetLoadBalancerName(context.TODO(), clusterName, svc)
+
 	svcNamespacedName := types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}
 	_, _, protocol := getPortsAndProtocol(svc.Spec.Ports)
 	scheme := cloud.SchemeInternal
 	sharedBackend := shareBackendService(svc)
 	sharedHealthCheck := !servicehelpers.RequestsOnlyLocalTraffic(svc)
 
-	g.sharedResourceLock.Lock()
-	defer g.sharedResourceLock.Unlock()
+	unlock := g.lockSharedResourcesIfCoarse()
+	defer unlock()
 
 	klog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): attempting delete of region internal address", loadBalancerName)
 	ensureAddressDeleted(g, loadBalancerName, g.region)
@@ -442,7 +443,7 @@ func (g *Cloud) ensureInternalLoadBalancerDeleted(clusterName, clusterID string,
 
 	hcName := makeHealthCheckName(loadBalancerName, clusterID, sharedHealthCheck)
 	klog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting health check %v and its firewall", loadBalancerName, hcName)
-	if err := g.teardownInternalHealthCheckAndFirewall(svc, hcName); err != nil {
+	if err := g.teardownInternalHealthCheckAndFirewall(svc, hcName, sharedHealthCheck); err != nil {
 		return err
 	}
 
@@ -480,7 +481,11 @@ func (g *Cloud) teardownInternalBackendService(bsName string) error {
 	return nil
 }
 
-func (g *Cloud) teardownInternalHealthCheckAndFirewall(svc *v1.Service, hcName string) error {
+func (g *Cloud) teardownInternalHealthCheckAndFirewall(svc *v1.Service, hcName string, shared bool) error {
+	hcFirewallName := makeHealthCheckFirewallNameFromHC(hcName)
+	defer g.lockHealthCheck(hcName, shared)()
+	defer g.lockFirewall(hcFirewallName, shared)()
+
 	if err := g.DeleteHealthCheck(hcName); err != nil {
 		if isNotFound(err) {
 			klog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check does not exist.", hcName)
@@ -494,7 +499,6 @@ func (g *Cloud) teardownInternalHealthCheckAndFirewall(svc *v1.Service, hcName s
 	}
 	klog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check deleted", hcName)
 
-	hcFirewallName := makeHealthCheckFirewallNameFromHC(hcName)
 	if err := ignoreNotFound(g.DeleteFirewall(hcFirewallName)); err != nil {
 		if isForbidden(err) && g.OnXPN() {
 			klog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): could not delete health check traffic firewall on XPN cluster. Raising Event.", hcName)
@@ -508,7 +512,9 @@ func (g *Cloud) teardownInternalHealthCheckAndFirewall(svc *v1.Service, hcName s
 	return nil
 }
 
-func (g *Cloud) ensureInternalFirewall(svc *v1.Service, fwName, fwDesc, destinationIP string, sourceRanges []string, portRanges []string, protocol v1.Protocol, nodes []*v1.Node, legacyFwName string) error {
+func (g *Cloud) ensureInternalFirewall(svc *v1.Service, fwName, fwDesc, destinationIP string, sourceRanges []string, portRanges []string, protocol v1.Protocol, nodes []*v1.Node, legacyFwName string, shared bool) error {
+	defer g.lockFirewall(fwName, shared)()
+
 	klog.V(2).Infof("ensureInternalFirewall(%v): checking existing firewall", fwName)
 	targetTags, err := g.GetNodeTags(nodeNames(nodes))
 	if err != nil {
@@ -595,18 +601,20 @@ func (g *Cloud) ensureInternalFirewalls(loadBalancerName, ipAddress, clusterID s
 		return err
 	}
 
-	err = g.ensureInternalFirewall(svc, MakeFirewallName(loadBalancerName), fwDesc, ipAddress, sourceRanges.StringSlice(), portRanges, protocol, nodes, loadBalancerName)
+	err = g.ensureInternalFirewall(svc, MakeFirewallName(loadBalancerName), fwDesc, ipAddress, sourceRanges.StringSlice(), portRanges, protocol, nodes, loadBalancerName, false)
 	if err != nil {
 		return err
 	}
 
 	// Second firewall is for health checking nodes / services
 	fwHCName := makeHealthCheckFirewallName(loadBalancerName, clusterID, sharedHealthCheck)
 	hcSrcRanges := L4LoadBalancerSrcRanges()
-	return g.ensureInternalFirewall(svc, fwHCName, "", "", hcSrcRanges, []string{healthCheckPort}, v1.ProtocolTCP, nodes, "")
+	return g.ensureInternalFirewall(svc, fwHCName, "", "", hcSrcRanges, []string{healthCheckPort}, v1.ProtocolTCP, nodes, "", sharedHealthCheck)
 }
 
 func (g *Cloud) ensureInternalHealthCheck(name string, svcName types.NamespacedName, shared bool, path string, port int32) (*compute.HealthCheck, error) {
+	defer g.lockHealthCheck(name, shared)()
+
 	klog.V(2).Infof("ensureInternalHealthCheck(%v, %v, %v): checking existing health check", name, path, port)
 	expectedHC := newInternalLBHealthCheck(name, svcName, shared, path, port)
 
@@ -646,6 +654,8 @@ func (g *Cloud) ensureInternalHealthCheck(name string, svcName types.NamespacedN
 }
 
 func (g *Cloud) ensureInternalInstanceGroup(name, zone string, nodes []*v1.Node, emptyZoneNodes []*v1.Node) (string, error) {
+	defer g.lockInstanceGroup(name, zone)()
+
 	klog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): checking group that it contains %v nodes [node names limited, total number of nodes: %d], the following nodes have empty string in the zone field and won't be deleted: %v", name, zone, loggableNodeNames(nodes), len(nodes), loggableNodeNames(emptyZoneNodes))
 	ig, err := g.GetInstanceGroup(name, zone)
 	if err != nil && !isNotFound(err) {
@@ -783,14 +793,24 @@ func (g *Cloud) ensureInternalInstanceGroupsDeleted(name string) error {
 	if !g.AlphaFeatureGate.Enabled(AlphaFeatureSkipIGsManagement) {
 		klog.V(2).Infof("ensureInternalInstanceGroupsDeleted(%v): attempting delete instance group in all %d zones", name, len(zones))
 		for _, z := range zones {
-			if err := g.DeleteInstanceGroup(name, z.Name); err != nil && !isNotFoundOrInUse(err) {
+			err := func() error {
+				defer g.lockInstanceGroup(name, z.Name)()
+
+				if err := g.DeleteInstanceGroup(name, z.Name); err != nil && !isNotFoundOrInUse(err) {
+					return err
+				}
+				return nil
+			}()
+			if err != nil {
 				return err
 			}
 		}
 	}
 	return nil
 }
 
+// Note: In the case of shared backend services,
+// concurrent updates are safely serialized by GCE's Optimistic Concurrency Control using resource fingerprints.
 func (g *Cloud) ensureInternalBackendService(name, description string, affinityType v1.ServiceAffinity, scheme cloud.LbScheme, protocol v1.Protocol, igLinks []string, hcLink string) error {
 	klog.V(2).Infof("ensureInternalBackendService(%v, %v, %v): checking existing backend service with %d groups", name, scheme, protocol, len(igLinks))
 	bs, err := g.GetRegionBackendService(name, g.region)