Clean backend pool node destination while reconciling security group (#9417) (#9422)

* Retain managed destination while reconciling security group

* Add docstring

Author: clnv

pkg/provider/azure.go

@@ -142,6 +142,7 @@ type Cloud struct {
 	pipCache azcache.Resource
 	// Add service lister to always get latest service
 	serviceLister corelisters.ServiceLister
+	nodeLister    corelisters.NodeLister
 	// node-sync-loop routine and service-reconcile routine should not update LoadBalancer at the same time
 	serviceReconcileLock sync.Mutex
 
@@ -722,6 +723,7 @@ func (az *Cloud) SetInformers(informerFactory informers.SharedInformerFactory) {
 	az.nodeInformerSynced = nodeInformer.HasSynced
 
 	az.serviceLister = informerFactory.Core().V1().Services().Lister()
+	az.nodeLister = informerFactory.Core().V1().Nodes().Lister()
 
 	az.setUpEndpointSlicesInformer(informerFactory)
 }

pkg/provider/azure_fakes.go

@@ -172,6 +172,7 @@ func GetTestCloud(ctrl *gomock.Controller) (az *Cloud) {
 		kubeClient := fake.NewSimpleClientset() // FIXME: inject kubeClient
 		informerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
 		az.serviceLister = informerFactory.Core().V1().Services().Lister()
+		az.nodeLister = informerFactory.Core().V1().Nodes().Lister()
 		informerFactory.Start(wait.NeverStop)
 		informerFactory.WaitForCacheSync(wait.NeverStop)
 	}

pkg/provider/azure_loadbalancer.go

@@ -3201,6 +3201,25 @@ func (az *Cloud) reconcileSecurityGroup(
 		}
 	}
 
+	{
+		// Retain all destinations that are managed by cloud-provider.
+		managedDestinations, err := az.listAvailableSecurityGroupDestinations(ctx)
+		if err != nil {
+			logger.Error(err, "Failed to list available security group destinations")
+			return nil, err
+		}
+
+		managedDestinations = append(managedDestinations, lbIPAddresses...)
+		managedDestinations = append(managedDestinations, additionalIPs...)
+		logger.Info("Retaining security group", "managed-destinations", managedDestinations)
+
+		ipv4Addresses, ipv6Addresses := iputil.GroupAddressesByFamily(managedDestinations)
+		if err := accessControl.RetainSecurityGroup(ipv4Addresses, ipv6Addresses); err != nil {
+			logger.Error(err, "Failed to retain security group")
+			return nil, err
+		}
+	}
+
 	rv, updated, err := accessControl.SecurityGroup()
 	if err != nil {
 		err = fmt.Errorf("unable to apply access control configuration to security group: %w", err)

pkg/provider/azure_loadbalancer_accesscontrol.go

@@ -106,3 +106,63 @@ func (az *Cloud) listSharedIPPortMapping(
 
 	return rv, nil
 }
+
+func (az *Cloud) listAvailableSecurityGroupDestinations(_ context.Context) ([]netip.Addr, error) {
+	services, err := az.serviceLister.List(labels.Everything())
+	if err != nil {
+		return nil, fmt.Errorf("list all services: %w", err)
+	}
+
+	nodes, err := az.nodeLister.List(labels.NewSelector())
+	if err != nil {
+		return nil, fmt.Errorf("list all nodes: %w", err)
+	}
+
+	var rv []netip.Addr
+	for _, svc := range services {
+		// Add additional public IPs
+		{
+			ips, err := loadbalancer.AdditionalPublicIPs(svc)
+			if err == nil {
+				rv = append(rv, ips...)
+			}
+		}
+
+		// Add ingress IPs
+		{
+			for _, ing := range svc.Status.LoadBalancer.Ingress {
+				ip, err := netip.ParseAddr(ing.IP)
+				if err == nil {
+					rv = append(rv, ip)
+				}
+			}
+		}
+	}
+
+	// Add backend node IPs
+	{
+		for _, node := range nodes {
+			if !az.isNodeManagedByCloudProvider(node) {
+				continue
+			}
+			for _, addr := range node.Status.Addresses {
+				if addr.Type != v1.NodeInternalIP {
+					continue
+				}
+				ip, err := netip.ParseAddr(addr.Address)
+				if err == nil {
+					rv = append(rv, ip)
+				}
+			}
+		}
+	}
+
+	return rv, nil
+}
+
+func (az *Cloud) isNodeManagedByCloudProvider(node *v1.Node) bool {
+	az.nodeCachesLock.Lock()
+	defer az.nodeCachesLock.Unlock()
+
+	return !az.unmanagedNodes.Has(node.ObjectMeta.Name)
+}