fix LoadBalancer in custom VPCs

kubeovn · Jul 30, 2021 · 80e5e2b · 80e5e2b
1 parent d7a514b
commit 80e5e2b
Show file tree

Hide file tree

Showing 5 changed files with 59 additions and 2 deletions.
diff --git a/dist/images/install-pre-1.16.sh b/dist/images/install-pre-1.16.sh
@@ -1222,6 +1222,10 @@ rules:
       - daemonsets
       - deployments
     verbs:
+      - create
+      - delete
+      - update
+      - patch
       - get
       - list
       - watch

diff --git a/dist/images/install.sh b/dist/images/install.sh
@@ -1264,6 +1264,10 @@ rules:
       - daemonsets
       - deployments
     verbs:
+      - create
+      - delete
+      - update
+      - patch
       - get
       - list
       - watch

diff --git a/pkg/controller/endpoint.go b/pkg/controller/endpoint.go
@@ -1,11 +1,14 @@
 package controller
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog"
@@ -117,16 +120,58 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
 		return nil
 	}
 
-	vpcName := svc.Annotations[util.VpcAnnotation]
+	pods, err := c.podsLister.Pods(namespace).List(labels.Set(svc.Spec.Selector).AsSelector())
+	if err != nil {
+		klog.Errorf("failed to get pods for service %s in namespace %s: %v", name, namespace, err)
+		return err
+	}
+
+	var vpcName string
+	for _, pod := range pods {
+		if len(pod.Annotations) == 0 {
+			continue
+		}
+
+		for _, subset := range ep.Subsets {
+			for _, addr := range subset.Addresses {
+				if addr.IP == pod.Status.PodIP {
+					if vpcName = pod.Annotations[util.LogicalRouterAnnotation]; vpcName != "" {
+						break
+					}
+				}
+			}
+			if vpcName != "" {
+				break
+			}
+		}
+		if vpcName != "" {
+			break
+		}
+	}
+
 	if vpcName == "" {
-		vpcName = util.DefaultVpc
+		if vpcName = svc.Annotations[util.VpcAnnotation]; vpcName == "" {
+			vpcName = util.DefaultVpc
+		}
 	}
+
 	vpc, err := c.vpcsLister.Get(vpcName)
 	if err != nil {
 		klog.Errorf("failed to get vpc %s of lb, %v", vpcName, err)
 		return err
 	}
 
+	if svcVpc := svc.Annotations[util.VpcAnnotation]; svcVpc != vpcName {
+		if svc.Annotations == nil {
+			svc.Annotations = make(map[string]string, 1)
+		}
+		svc.Annotations[util.VpcAnnotation] = vpcName
+		if svc, err = c.config.KubeClient.CoreV1().Services(namespace).Update(context.Background(), svc, metav1.UpdateOptions{}); err != nil {
+			klog.Errorf("failed to update service %s/%s: %v", namespace, svc.Name, err)
+			return err
+		}
+	}
+
 	tcpLb, udpLb := vpc.Status.TcpLoadBalancer, vpc.Status.UdpLoadBalancer
 	if svc.Spec.SessionAffinity == v1.ServiceAffinityClientIP {
 		tcpLb, udpLb = vpc.Status.TcpSessionLoadBalancer, vpc.Status.UdpSessionLoadBalancer

diff --git a/pkg/controller/pod.go b/pkg/controller/pod.go
@@ -416,6 +416,9 @@ func (c *Controller) handleAddPod(key string) error {
 		if pod.Annotations[util.PodNicAnnotation] == "" {
 			pod.Annotations[util.PodNicAnnotation] = c.config.PodNicType
 		}
+		if subnet.Spec.Vlan == "" && subnet.Spec.Vpc != "" {
+			pod.Annotations[fmt.Sprintf(util.LogicalRouterAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.Vpc
+		}
 
 		if err := util.ValidatePodCidr(podNet.Subnet.Spec.CIDRBlock, ipStr); err != nil {
 			klog.Errorf("validate pod %s/%s failed, %v", namespace, name, err)

diff --git a/pkg/util/const.go b/pkg/util/const.go
@@ -40,6 +40,7 @@ const (
 	GatewayAnnotationTemplate       = "%s.kubernetes.io/gateway"
 	IpPoolAnnotationTemplate        = "%s.kubernetes.io/ip_pool"
 	LogicalSwitchAnnotationTemplate = "%s.kubernetes.io/logical_switch"
+	LogicalRouterAnnotationTemplate = "%s.kubernetes.io/logical_router"
 	VlanIdAnnotationTemplate        = "%s.kubernetes.io/vlan_id"
 	NetworkTypeTemplate             = "%s.kubernetes.io/network_type"
 	IngressRateAnnotationTemplate   = "%s.kubernetes.io/ingress_rate"