fix ipv6 service ip not added to ovn lb vips due to pod cache not synced

Signed-off-by: zhangzujian <[email protected]>

pkg/controller/endpoint.go

@@ -106,14 +106,15 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
 	defer func() { _ = c.epKeyMutex.UnlockKey(key) }()
 	klog.Infof("handle update endpoint %s", key)
 
-	ep, err := c.endpointsLister.Endpoints(namespace).Get(name)
+	cachedEndpoint, err := c.endpointsLister.Endpoints(namespace).Get(name)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil
 		}
 		klog.Error(err)
 		return err
 	}
+	ep := cachedEndpoint.DeepCopy()
 
 	cachedService, err := c.servicesLister.Services(namespace).Get(name)
 	if err != nil {
@@ -153,6 +154,30 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
 		klog.Errorf("failed to get pods for service %s in namespace %s: %v", name, namespace, err)
 		return err
 	}
+	for i, pod := range pods {
+		if pod.Status.PodIP != "" || len(pod.Status.PodIPs) != 0 {
+			continue
+		}
+
+		for _, subset := range ep.Subsets {
+			for _, addr := range subset.Addresses {
+				if addr.TargetRef == nil || addr.TargetRef.Kind != "Pod" || addr.TargetRef.Name != pod.Name {
+					continue
+				}
+
+				p, err := c.config.KubeClient.CoreV1().Pods(pod.Namespace).Get(context.Background(), pod.Name, metav1.GetOptions{})
+				if err != nil {
+					klog.Errorf("failed to get pod %s/%s: %v", pod.Namespace, pod.Name, err)
+					return err
+				}
+				pods[i] = p.DeepCopy()
+				break
+			}
+			if pods[i] != pod {
+				break
+			}
+		}
+	}
 
 	vpcName, subnetName = c.getVpcSubnetName(pods, ep, svc)
 

pkg/controller/gc.go

@@ -495,12 +495,7 @@ func (c *Controller) gcLoadBalancer() error {
 	)
 
 	for _, svc := range svcs {
-		ips := util.ServiceClusterIPs(*svc)
-		if v, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {
-			ips = strings.Split(v, ",")
-		}
-
-		for _, ip := range ips {
+		for _, ip := range getVipIps(svc) {
 			for _, port := range svc.Spec.Ports {
 				vip := util.JoinHostPort(ip, port.Port)
 				switch port.Protocol {

pkg/controller/service.go

@@ -382,11 +382,9 @@ func (c *Controller) handleUpdateService(key string) error {
 				return err
 			}
 
-			if !needUpdateEndpointQueue {
-				if _, ok := lb.Vips[vip]; !ok {
-					klog.Infof("add vip %s to LB %s", vip, lbName)
-					needUpdateEndpointQueue = true
-				}
+			if _, ok := lb.Vips[vip]; !ok {
+				klog.Infof("add vip %s to LB %s", vip, lbName)
+				needUpdateEndpointQueue = true
 			}
 		}
 		for vip := range lb.Vips {

test/e2e/kube-ovn/service/service.go

@@ -137,15 +137,18 @@ var _ = framework.Describe("[group:service]", func() {
 		selector := map[string]string{"app": "svc-dual"}
 		service := framework.MakeService(serviceName, corev1.ServiceTypeClusterIP, nil, selector, ports, corev1.ServiceAffinityNone)
 		service.Namespace = namespaceName
+		service.Spec.IPFamilyPolicy = ptr.To(corev1.IPFamilyPolicyRequireDualStack)
 		service = serviceClient.CreateSync(service, func(s *corev1.Service) (bool, error) {
-			return len(s.Spec.ClusterIPs) != 0, nil
-		}, "cluster ips are not empty")
+			return len(util.ServiceClusterIPs(*s)) == 2, nil
+		}, "both ipv4 and ipv6 cluster ips are allocated")
 		v6ClusterIP := service.Spec.ClusterIPs[1]
 		originService := service.DeepCopy()
+		framework.Logf("created service %s with cluster ips %s", serviceName, strings.Join(util.ServiceClusterIPs(*service), ","))
 
 		ginkgo.By("Creating pod " + podName)
 		podBackend := framework.MakePod(namespaceName, podName, selector, nil, framework.PauseImage, nil, nil)
-		_ = podClient.CreateSync(podBackend)
+		podBackend = podClient.CreateSync(podBackend)
+		framework.Logf("created pod %s with ips %s", podName, strings.Join(util.PodIPs(*podBackend), ","))
 
 		checkContainsClusterIP := func(v6ClusterIP string, isContain bool) {
 			ginkgo.GinkgoHelper()