21056235309

Committed 16 Jan 2026 05:02AM UTC coverage: 21.401% (-0.02%) from 21.42%

Build # 21056235309

Build Type

push

github

Committed by

web-flow

Commit Message

1. fix u2o cause metallb underlay err (#6173)

* 1. fix u2o cause metallb underlay err
2. fix underlay metallb ipv6 err

Signed-off-by: clyi <clyi@alauda.io>

Run Details

0 of 62 new or added lines in 3 files covered. (0.0%)

2 existing lines in 2 files now uncovered.

10635 of 49695 relevant lines covered (21.4%)

0.5 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.2

/pkg/controller/endpoint_slice.go

package controller

import (
        "context"
        "fmt"
        "strings"
        "time"

        v1 "k8s.io/api/core/v1"
        discoveryv1 "k8s.io/api/discovery/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/v2"

        kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
        "github.com/kubeovn/kube-ovn/pkg/ovs"
        "github.com/kubeovn/kube-ovn/pkg/util"
)

func findServiceKey(endpointSlice *discoveryv1.EndpointSlice) string {
        if endpointSlice != nil && endpointSlice.Labels != nil && endpointSlice.Labels[discoveryv1.LabelServiceName] != "" {
                return endpointSlice.Namespace + "/" + endpointSlice.Labels[discoveryv1.LabelServiceName]
        }
        return ""
}

func (c *Controller) enqueueAddEndpointSlice(obj any) {
        key := findServiceKey(obj.(*discoveryv1.EndpointSlice))
        if key != "" {
                klog.V(3).Infof("enqueue add endpointSlice %s", key)
                c.addOrUpdateEndpointSliceQueue.Add(key)
        }
}

func (c *Controller) enqueueUpdateEndpointSlice(oldObj, newObj any) {
        oldEndpointSlice := oldObj.(*discoveryv1.EndpointSlice)
        newEndpointSlice := newObj.(*discoveryv1.EndpointSlice)
        if oldEndpointSlice.ResourceVersion == newEndpointSlice.ResourceVersion {
                return
        }

        if len(oldEndpointSlice.Endpoints) == 0 && len(newEndpointSlice.Endpoints) == 0 {
                return
        }

        key := findServiceKey(newEndpointSlice)
        if key != "" {
                klog.V(3).Infof("enqueue update endpointSlice for service %s", key)
                c.addOrUpdateEndpointSliceQueue.Add(key)
        }
}

func (c *Controller) handleUpdateEndpointSlice(key string) error {
        namespace, name, err := cache.SplitMetaNamespaceKey(key)
        if err != nil {
                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
                return nil
        }

        c.epKeyMutex.LockKey(key)
        defer func() { _ = c.epKeyMutex.UnlockKey(key) }()
        klog.Infof("handle update endpointSlice for service %s", key)

        endpointSlices, err := c.endpointSlicesLister.EndpointSlices(namespace).List(labels.Set(map[string]string{discoveryv1.LabelServiceName: name}).AsSelector())
        if err != nil {
                if errors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }

        cachedService, err := c.servicesLister.Services(namespace).Get(name)
        if err != nil {
                if errors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        svc := cachedService.DeepCopy()

        var (
                pods                     []*v1.Pod
                lbVips                   []string
                vip, vpcName, subnetName string
                ok                       bool
                ignoreHealthCheck        = true
                isPreferLocalBackend     = false
        )

        if vip, ok = svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {
                lbVips = []string{vip}

                for _, endpointSlice := range endpointSlices {
                        for _, endpoint := range endpointSlice.Endpoints {
                                if util.CheckProtocol(vip) == kubeovnv1.ProtocolIPv4 &&
                                        endpoint.TargetRef.Name != "" {
                                        ignoreHealthCheck = false
                                }
                        }
                }
        } else if lbVips = util.ServiceClusterIPs(*svc); len(lbVips) == 0 {
                return nil
        }

        if c.config.EnableLb && c.config.EnableOVNLBPreferLocal {
                if svc.Spec.Type == v1.ServiceTypeLoadBalancer && svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {
                        if len(svc.Status.LoadBalancer.Ingress) > 0 {
                                for _, ingress := range svc.Status.LoadBalancer.Ingress {
                                        if ingress.IP != "" {
                                                lbVips = append(lbVips, ingress.IP)
                                        }
                                }
                        }
                        isPreferLocalBackend = true
                } else if svc.Spec.Type == v1.ServiceTypeClusterIP && svc.Spec.InternalTrafficPolicy != nil && *svc.Spec.InternalTrafficPolicy == v1.ServiceInternalTrafficPolicyLocal {
                        isPreferLocalBackend = true
                }
        }

        if pods, err = c.podsLister.Pods(namespace).List(labels.Set(svc.Spec.Selector).AsSelector()); err != nil {
                klog.Errorf("failed to get pods for service %s in namespace %s: %v", name, namespace, err)
                return err
        }
        vpcName, subnetName = c.getVpcSubnetName(pods, endpointSlices, svc)

        var (
                vpc    *kubeovnv1.Vpc
                svcVpc string
        )

        if vpc, err = c.vpcsLister.Get(vpcName); err != nil {
                klog.Errorf("failed to get vpc %s, %v", vpcName, err)
                return err
        }

        tcpLb, udpLb, sctpLb := vpc.Status.TCPLoadBalancer, vpc.Status.UDPLoadBalancer, vpc.Status.SctpLoadBalancer
        oldTCPLb, oldUDPLb, oldSctpLb := vpc.Status.TCPSessionLoadBalancer, vpc.Status.UDPSessionLoadBalancer, vpc.Status.SctpSessionLoadBalancer
        if svc.Spec.SessionAffinity == v1.ServiceAffinityClientIP {
                tcpLb, udpLb, sctpLb, oldTCPLb, oldUDPLb, oldSctpLb = oldTCPLb, oldUDPLb, oldSctpLb, tcpLb, udpLb, sctpLb
        }

        for _, lbVip := range lbVips {
                for _, port := range svc.Spec.Ports {
                        var lb, oldLb string
                        switch port.Protocol {
                        case v1.ProtocolTCP:
                                lb, oldLb = tcpLb, oldTCPLb
                        case v1.ProtocolUDP:
                                lb, oldLb = udpLb, oldUDPLb
                        case v1.ProtocolSCTP:
                                lb, oldLb = sctpLb, oldSctpLb
                        }

                        var (
                                vip, checkIP             string
                                backends                 []string
                                ipPortMapping, externals map[string]string
                        )

                        if !ignoreHealthCheck {
                                if checkIP, err = c.getHealthCheckVip(subnetName, lbVip); err != nil {
                                        klog.Error(err)
                                        return err
                                }
                                externals = map[string]string{
                                        util.SwitchLBRuleSubnet: subnetName,
                                }
                        }

                        if isPreferLocalBackend {
                                // only use the ipportmapping's lsp to ip map when the backend is local
                                checkIP = util.MasqueradeCheckIP
                        }
                        isGenIPPortMapping := !ignoreHealthCheck || isPreferLocalBackend
                        ipPortMapping, backends = getIPPortMappingBackend(endpointSlices, port, lbVip, checkIP, isGenIPPortMapping)
                        // for performance reason delete lb with no backends
                        if len(backends) != 0 {
                                vip = util.JoinHostPort(lbVip, port.Port)
                                klog.Infof("add vip endpoint %s, backends %v to LB %s", vip, backends, lb)
                                if err = c.OVNNbClient.LoadBalancerAddVip(lb, vip, backends...); err != nil {
                                        klog.Errorf("failed to add vip %s with backends %s to LB %s: %v", lbVip, backends, lb, err)
                                        return err
                                }

                                if isPreferLocalBackend && len(ipPortMapping) != 0 {
                                        if err = c.OVNNbClient.LoadBalancerUpdateIPPortMapping(lb, vip, ipPortMapping); err != nil {
                                                klog.Errorf("failed to update ip port mapping %s for vip %s to LB %s: %v", ipPortMapping, vip, lb, err)
                                                return err
                                        }
                                }

                                if !ignoreHealthCheck && len(ipPortMapping) != 0 {
                                        klog.Infof("add health check ip port mapping %v to LB %s", ipPortMapping, lb)
                                        if err = c.OVNNbClient.LoadBalancerAddHealthCheck(lb, vip, ignoreHealthCheck, ipPortMapping, externals); err != nil {
                                                klog.Errorf("failed to add health check for vip %s with ip port mapping %s to LB %s: %v", lbVip, ipPortMapping, lb, err)
                                                return err
                                        }
                                }
                        } else {
                                vip = util.JoinHostPort(lbVip, port.Port)
                                klog.V(3).Infof("delete vip endpoint %s from LB %s", vip, lb)
                                if err = c.OVNNbClient.LoadBalancerDeleteVip(lb, vip, true); err != nil {
                                        klog.Errorf("failed to delete vip endpoint %s from LB %s: %v", vip, lb, err)
                                        return err
                                }

                                klog.V(3).Infof("delete vip endpoint %s from old LB %s", vip, oldLb)
                                if err = c.OVNNbClient.LoadBalancerDeleteVip(oldLb, vip, true); err != nil {
                                        klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oldLb, err)
                                        return err
                                }

                                if c.config.EnableOVNLBPreferLocal {
                                        if err := c.OVNNbClient.LoadBalancerDeleteIPPortMapping(lb, vip); err != nil {
                                                klog.Errorf("failed to delete ip port mapping for vip %s from LB %s: %v", vip, lb, err)
                                                return err
                                        }
                                        if err := c.OVNNbClient.LoadBalancerDeleteIPPortMapping(oldLb, vip); err != nil {
                                                klog.Errorf("failed to delete ip port mapping for vip %s from LB %s: %v", vip, lb, err)
                                                return err
                                        }
                                }
                        }
                }
        }

        if svcVpc = svc.Annotations[util.VpcAnnotation]; svcVpc != vpcName {
                patch := util.KVPatch{util.VpcAnnotation: vpcName}
                if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Services(namespace), svc.Name, patch); err != nil {
                        klog.Errorf("failed to patch service %s: %v", key, err)
                        return err
                }
        }

        return nil
}

func (c *Controller) getVpcSubnetName(pods []*v1.Pod, endpointSlices []*discoveryv1.EndpointSlice, service *v1.Service) (string, string) {
        var (
                vpcName    string
                subnetName string
        )

        for _, pod := range pods {
                if len(pod.Annotations) == 0 {
                        continue
                }
                if subnetName == "" {
                        subnetName = pod.Annotations[util.LogicalSwitchAnnotation]
                }

        LOOP:
                for _, endpointSlice := range endpointSlices {
                        for _, endpoint := range endpointSlice.Endpoints {
                                for _, addr := range endpoint.Addresses {
                                        for _, podIP := range pod.Status.PodIPs {
                                                if addr == podIP.IP {
                                                        if vpcName == "" {
                                                                vpcName = pod.Annotations[util.LogicalRouterAnnotation]
                                                        }
                                                        if vpcName != "" {
                                                                break LOOP
                                                        }
                                                }
                                        }
                                }
                        }
                }

                if vpcName != "" && subnetName != "" {
                        break
                }
        }

        if vpcName == "" {
                if vpcName = service.Annotations[util.VpcAnnotation]; vpcName == "" {
                        vpcName = c.config.ClusterRouter
                }
        }

        if subnetName == "" {
                subnetName = util.DefaultSubnet
        }

        return vpcName, subnetName
}

// getHealthCheckVip get health check vip for load balancer, the vip name is the subnet name
// the vip is used to check the health of the backend pod
func (c *Controller) getHealthCheckVip(subnetName, lbVip string) (string, error) {
        var (
                needCreateHealthCheckVip bool
                checkVip                 *kubeovnv1.Vip
                checkIP                  string
                err                      error
        )
        vipName := subnetName
        checkVip, err = c.virtualIpsLister.Get(vipName)
        if err != nil {
                if errors.IsNotFound(err) {
                        needCreateHealthCheckVip = true
                } else {
                        klog.Errorf("failed to get health check vip %s, %v", vipName, err)
                        return "", err
                }
        }
        if needCreateHealthCheckVip {
                vip := &kubeovnv1.Vip{
                        ObjectMeta: metav1.ObjectMeta{
                                Name: vipName,
                        },
                        Spec: kubeovnv1.VipSpec{
                                Subnet: subnetName,
                        },
                }
                if _, err = c.config.KubeOvnClient.KubeovnV1().Vips().Create(context.Background(), vip, metav1.CreateOptions{}); err != nil {
                        klog.Errorf("failed to create health check vip %s, %v", vipName, err)
                        return "", err
                }

                // wait for vip created
                time.Sleep(1 * time.Second)
                checkVip, err = c.virtualIpsLister.Get(vipName)
                if err != nil {
                        klog.Errorf("failed to get health check vip %s, %v", vipName, err)
                        return "", err
                }
        }

        if checkVip.Status.V4ip == "" && checkVip.Status.V6ip == "" {
                err = fmt.Errorf("vip %s is not ready", vipName)
                klog.Error(err)
                return "", err
        }

        switch util.CheckProtocol(lbVip) {
        case kubeovnv1.ProtocolIPv4:
                checkIP = checkVip.Status.V4ip
        case kubeovnv1.ProtocolIPv6:
                checkIP = checkVip.Status.V6ip
        }
        if checkIP == "" {
                err = fmt.Errorf("failed to get health check vip subnet %s", vipName)
                klog.Error(err)
                return "", err
        }

        return checkIP, nil
}

func getIPPortMappingBackend(endpointSlices []*discoveryv1.EndpointSlice, servicePort v1.ServicePort, serviceIP, checkVip string, isGenIPPortMapping bool) (map[string]string, []string) {
        var (
                ipPortMapping = map[string]string{}
                backends      = []string{}
                protocol      = util.CheckProtocol(serviceIP)
        )

        for _, endpointSlice := range endpointSlices {
                var targetPort int32
                for _, port := range endpointSlice.Ports {
                        if port.Name != nil && *port.Name == servicePort.Name {
                                targetPort = *port.Port
                                break
                        }
                }
                if targetPort == 0 {
                        continue
                }

                for _, endpoint := range endpointSlice.Endpoints {
                        if isGenIPPortMapping && endpoint.TargetRef.Name != "" {
                                lspName := getEndpointTargetLSP(endpoint.TargetRef.Name, endpoint.TargetRef.Namespace, util.OvnProvider)
                                for _, address := range endpoint.Addresses {
                                        key := address
                                        if util.CheckProtocol(address) == kubeovnv1.ProtocolIPv6 {
                                                key = fmt.Sprintf("[%s]", address)
                                        }
                                        ipPortMapping[key] = fmt.Sprintf(util.HealthCheckNamedVipTemplate, lspName, checkVip)
                                }
                        }
                }

                for _, endpoint := range endpointSlice.Endpoints {
                        if !endpointReady(endpoint) {
                                continue
                        }

                        for _, address := range endpoint.Addresses {
                                if util.CheckProtocol(address) == protocol {
                                        backends = append(backends, util.JoinHostPort(address, targetPort))
                                }
                        }
                }
        }

        return ipPortMapping, backends
}

func endpointReady(endpoint discoveryv1.Endpoint) bool {
        return endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready
}

// getEndpointTargetLSP returns the name of the LSP for a given target/namespace.
// A custom provider can be specified if the LSP is within a subnet that doesn't use
// the default "ovn" provider.
func getEndpointTargetLSP(target, namespace, provider string) string {
        // This pod seems to be a VM launcher pod, but we do not use the same syntax for the LSP
        // of normal pods and for VM pods. We need to retrieve the real name of the VM from
        // the pod's name to compute the LSP.
        if strings.HasPrefix(target, util.VMLauncherPrefix) {
                target = getVMNameFromLauncherPod(target)
        }

        return ovs.PodNameToPortName(target, namespace, provider)
}

// getVMNameFromLauncherPod returns the name of a VirtualMachine from the name of its launcher pod (virt-launcher)
func getVMNameFromLauncherPod(podName string) string {
        // Remove the VM launcher pod prefix
        vmName := strings.TrimPrefix(podName, util.VMLauncherPrefix)

        // Remove the ID of the pod
        slice := strings.Split(vmName, "-")
        if len(slice) > 0 {
                vmName = strings.Join(slice[:len(slice)-1], "-")
        }

        return vmName
}

1	package controller
2
3	import (
4	"context"
5	"fmt"
6	"strings"
7	"time"
8
9	v1 "k8s.io/api/core/v1"
10	discoveryv1 "k8s.io/api/discovery/v1"
11	"k8s.io/apimachinery/pkg/api/errors"
12	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
13	"k8s.io/apimachinery/pkg/labels"
14	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
15	"k8s.io/client-go/tools/cache"
16	"k8s.io/klog/v2"
17
18	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
19	"github.com/kubeovn/kube-ovn/pkg/ovs"
20	"github.com/kubeovn/kube-ovn/pkg/util"
21	)
22
23	func findServiceKey(endpointSlice *discoveryv1.EndpointSlice) string {	2✔
24	if endpointSlice != nil && endpointSlice.Labels != nil && endpointSlice.Labels[discoveryv1.LabelServiceName] != "" {	4✔
25	return endpointSlice.Namespace + "/" + endpointSlice.Labels[discoveryv1.LabelServiceName]	2✔
26	}	2✔
27	return ""	2✔
28	}
29
30	func (c *Controller) enqueueAddEndpointSlice(obj any) {	×
31	key := findServiceKey(obj.(*discoveryv1.EndpointSlice))	×
32	if key != "" {	×
33	klog.V(3).Infof("enqueue add endpointSlice %s", key)	×
34	c.addOrUpdateEndpointSliceQueue.Add(key)	×
35	}	×
36	}
37
38	func (c *Controller) enqueueUpdateEndpointSlice(oldObj, newObj any) {	×
39	oldEndpointSlice := oldObj.(*discoveryv1.EndpointSlice)	×
40	newEndpointSlice := newObj.(*discoveryv1.EndpointSlice)	×
41	if oldEndpointSlice.ResourceVersion == newEndpointSlice.ResourceVersion {	×
42	return	×
43	}	×
44
45	if len(oldEndpointSlice.Endpoints) == 0 && len(newEndpointSlice.Endpoints) == 0 {	×
46	return	×
47	}	×
48
49	key := findServiceKey(newEndpointSlice)	×
50	if key != "" {	×
51	klog.V(3).Infof("enqueue update endpointSlice for service %s", key)	×
52	c.addOrUpdateEndpointSliceQueue.Add(key)	×
53	}	×
54	}
55
56	func (c *Controller) handleUpdateEndpointSlice(key string) error {	×
57	namespace, name, err := cache.SplitMetaNamespaceKey(key)	×
58	if err != nil {	×
59	utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))	×
60	return nil	×
61	}	×
62
63	c.epKeyMutex.LockKey(key)	×
64	defer func() { _ = c.epKeyMutex.UnlockKey(key) }()	×
65	klog.Infof("handle update endpointSlice for service %s", key)	×
66		×
67	endpointSlices, err := c.endpointSlicesLister.EndpointSlices(namespace).List(labels.Set(map[string]string{discoveryv1.LabelServiceName: name}).AsSelector())	×
68	if err != nil {	×
69	if errors.IsNotFound(err) {	×
70	return nil	×
71	}	×
72	klog.Error(err)	×
73	return err	×
74	}
75
76	cachedService, err := c.servicesLister.Services(namespace).Get(name)	×
77	if err != nil {	×
78	if errors.IsNotFound(err) {	×
79	return nil	×
80	}	×
81	klog.Error(err)	×
82	return err	×
83	}
84	svc := cachedService.DeepCopy()	×
85		×
86	var (	×
87	pods []*v1.Pod	×
88	lbVips []string	×
89	vip, vpcName, subnetName string	×
90	ok bool	×
91	ignoreHealthCheck = true	×
92	isPreferLocalBackend = false	×
93	)	×
94		×
95	if vip, ok = svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {	×
96	lbVips = []string{vip}	×
97		×
98	for _, endpointSlice := range endpointSlices {	×
99	for _, endpoint := range endpointSlice.Endpoints {	×
100	if util.CheckProtocol(vip) == kubeovnv1.ProtocolIPv4 &&	×
101	endpoint.TargetRef.Name != "" {	×
102	ignoreHealthCheck = false	×
103	}	×
104	}
105	}
106	} else if lbVips = util.ServiceClusterIPs(*svc); len(lbVips) == 0 {	×
107	return nil	×
108	}	×
109
110	if c.config.EnableLb && c.config.EnableOVNLBPreferLocal {	×
111	if svc.Spec.Type == v1.ServiceTypeLoadBalancer && svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {	×
112	if len(svc.Status.LoadBalancer.Ingress) > 0 {	×
113	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
114	if ingress.IP != "" {	×
115	lbVips = append(lbVips, ingress.IP)	×
116	}	×
117	}
118	}
119	isPreferLocalBackend = true	×
120	} else if svc.Spec.Type == v1.ServiceTypeClusterIP && svc.Spec.InternalTrafficPolicy != nil && *svc.Spec.InternalTrafficPolicy == v1.ServiceInternalTrafficPolicyLocal {	×
121	isPreferLocalBackend = true	×
122	}	×
123	}
124
125	if pods, err = c.podsLister.Pods(namespace).List(labels.Set(svc.Spec.Selector).AsSelector()); err != nil {	×
126	klog.Errorf("failed to get pods for service %s in namespace %s: %v", name, namespace, err)	×
127	return err	×
128	}	×
129	vpcName, subnetName = c.getVpcSubnetName(pods, endpointSlices, svc)	×
130		×
131	var (	×
132	vpc *kubeovnv1.Vpc	×
133	svcVpc string	×
134	)	×
135		×
136	if vpc, err = c.vpcsLister.Get(vpcName); err != nil {	×
137	klog.Errorf("failed to get vpc %s, %v", vpcName, err)	×
138	return err	×
139	}	×
140
141	tcpLb, udpLb, sctpLb := vpc.Status.TCPLoadBalancer, vpc.Status.UDPLoadBalancer, vpc.Status.SctpLoadBalancer	×
142	oldTCPLb, oldUDPLb, oldSctpLb := vpc.Status.TCPSessionLoadBalancer, vpc.Status.UDPSessionLoadBalancer, vpc.Status.SctpSessionLoadBalancer	×
143	if svc.Spec.SessionAffinity == v1.ServiceAffinityClientIP {	×
144	tcpLb, udpLb, sctpLb, oldTCPLb, oldUDPLb, oldSctpLb = oldTCPLb, oldUDPLb, oldSctpLb, tcpLb, udpLb, sctpLb	×
145	}	×
146
147	for _, lbVip := range lbVips {	×
148	for _, port := range svc.Spec.Ports {	×
149	var lb, oldLb string	×
150	switch port.Protocol {	×
151	case v1.ProtocolTCP:	×
152	lb, oldLb = tcpLb, oldTCPLb	×
153	case v1.ProtocolUDP:	×
154	lb, oldLb = udpLb, oldUDPLb	×
155	case v1.ProtocolSCTP:	×
156	lb, oldLb = sctpLb, oldSctpLb	×
157	}
158
159	var (	×
160	vip, checkIP string	×
161	backends []string	×
162	ipPortMapping, externals map[string]string	×
163	)	×
164		×
165	if !ignoreHealthCheck {	×
166	if checkIP, err = c.getHealthCheckVip(subnetName, lbVip); err != nil {	×
167	klog.Error(err)	×
168	return err	×
169	}	×
170	externals = map[string]string{	×
171	util.SwitchLBRuleSubnet: subnetName,	×
172	}	×
173	}
174
175	if isPreferLocalBackend {	×
176	// only use the ipportmapping's lsp to ip map when the backend is local	×
177	checkIP = util.MasqueradeCheckIP	×
178	}	×
179	isGenIPPortMapping := !ignoreHealthCheck \|\| isPreferLocalBackend	×
180	ipPortMapping, backends = getIPPortMappingBackend(endpointSlices, port, lbVip, checkIP, isGenIPPortMapping)	×
181	// for performance reason delete lb with no backends	×
182	if len(backends) != 0 {	×
183	vip = util.JoinHostPort(lbVip, port.Port)	×
184	klog.Infof("add vip endpoint %s, backends %v to LB %s", vip, backends, lb)	×
185	if err = c.OVNNbClient.LoadBalancerAddVip(lb, vip, backends...); err != nil {	×
186	klog.Errorf("failed to add vip %s with backends %s to LB %s: %v", lbVip, backends, lb, err)	×
187	return err	×
188	}	×
189
190	if isPreferLocalBackend && len(ipPortMapping) != 0 {	×
191	if err = c.OVNNbClient.LoadBalancerUpdateIPPortMapping(lb, vip, ipPortMapping); err != nil {	×
192	klog.Errorf("failed to update ip port mapping %s for vip %s to LB %s: %v", ipPortMapping, vip, lb, err)	×
193	return err	×
194	}	×
195	}
196
197	if !ignoreHealthCheck && len(ipPortMapping) != 0 {	×
198	klog.Infof("add health check ip port mapping %v to LB %s", ipPortMapping, lb)	×
199	if err = c.OVNNbClient.LoadBalancerAddHealthCheck(lb, vip, ignoreHealthCheck, ipPortMapping, externals); err != nil {	×
200	klog.Errorf("failed to add health check for vip %s with ip port mapping %s to LB %s: %v", lbVip, ipPortMapping, lb, err)	×
201	return err	×
202	}	×
203	}
204	} else {	×
205	vip = util.JoinHostPort(lbVip, port.Port)	×
206	klog.V(3).Infof("delete vip endpoint %s from LB %s", vip, lb)	×
207	if err = c.OVNNbClient.LoadBalancerDeleteVip(lb, vip, true); err != nil {	×
208	klog.Errorf("failed to delete vip endpoint %s from LB %s: %v", vip, lb, err)	×
209	return err	×
210	}	×
211
212	klog.V(3).Infof("delete vip endpoint %s from old LB %s", vip, oldLb)	×
213	if err = c.OVNNbClient.LoadBalancerDeleteVip(oldLb, vip, true); err != nil {	×
214	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oldLb, err)	×
215	return err	×
216	}	×
217
218	if c.config.EnableOVNLBPreferLocal {	×
219	if err := c.OVNNbClient.LoadBalancerDeleteIPPortMapping(lb, vip); err != nil {	×
220	klog.Errorf("failed to delete ip port mapping for vip %s from LB %s: %v", vip, lb, err)	×
221	return err	×
222	}	×
223	if err := c.OVNNbClient.LoadBalancerDeleteIPPortMapping(oldLb, vip); err != nil {	×
224	klog.Errorf("failed to delete ip port mapping for vip %s from LB %s: %v", vip, lb, err)	×
225	return err	×
226	}	×
227	}
228	}
229	}
230	}
231
232	if svcVpc = svc.Annotations[util.VpcAnnotation]; svcVpc != vpcName {	×
233	patch := util.KVPatch{util.VpcAnnotation: vpcName}	×
234	if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Services(namespace), svc.Name, patch); err != nil {	×
235	klog.Errorf("failed to patch service %s: %v", key, err)	×
236	return err	×
237	}	×
238	}
239
240	return nil	×
241	}
242
243	func (c Controller) getVpcSubnetName(pods []v1.Pod, endpointSlices []discoveryv1.EndpointSlice, service v1.Service) (string, string) {	×
244	var (	×
245	vpcName string	×
246	subnetName string	×
247	)	×
248		×
249	for _, pod := range pods {	×
250	if len(pod.Annotations) == 0 {	×
251	continue	×
252	}
253	if subnetName == "" {	×
254	subnetName = pod.Annotations[util.LogicalSwitchAnnotation]	×
255	}	×
256
257	LOOP:	×
258	for _, endpointSlice := range endpointSlices {	×
259	for _, endpoint := range endpointSlice.Endpoints {	×
260	for _, addr := range endpoint.Addresses {	×
261	for _, podIP := range pod.Status.PodIPs {	×
262	if addr == podIP.IP {	×
263	if vpcName == "" {	×
264	vpcName = pod.Annotations[util.LogicalRouterAnnotation]	×
265	}	×
266	if vpcName != "" {	×
267	break LOOP	×
268	}
269	}
270	}
271	}
272	}
273	}
274
275	if vpcName != "" && subnetName != "" {	×
276	break	×
277	}
278	}
279
280	if vpcName == "" {	×
281	if vpcName = service.Annotations[util.VpcAnnotation]; vpcName == "" {	×
282	vpcName = c.config.ClusterRouter	×
283	}	×
284	}
285
286	if subnetName == "" {	×
287	subnetName = util.DefaultSubnet	×
288	}	×
289
290	return vpcName, subnetName	×
291	}
292
293	// getHealthCheckVip get health check vip for load balancer, the vip name is the subnet name
294	// the vip is used to check the health of the backend pod
295	func (c *Controller) getHealthCheckVip(subnetName, lbVip string) (string, error) {	×
296	var (	×
297	needCreateHealthCheckVip bool	×
298	checkVip *kubeovnv1.Vip	×
299	checkIP string	×
300	err error	×
301	)	×
302	vipName := subnetName	×
303	checkVip, err = c.virtualIpsLister.Get(vipName)	×
304	if err != nil {	×
305	if errors.IsNotFound(err) {	×
306	needCreateHealthCheckVip = true	×
307	} else {	×
308	klog.Errorf("failed to get health check vip %s, %v", vipName, err)	×
309	return "", err	×
310	}	×
311	}
312	if needCreateHealthCheckVip {	×
313	vip := &kubeovnv1.Vip{	×
314	ObjectMeta: metav1.ObjectMeta{	×
315	Name: vipName,	×
316	},	×
317	Spec: kubeovnv1.VipSpec{	×
318	Subnet: subnetName,	×
319	},	×
320	}	×
321	if _, err = c.config.KubeOvnClient.KubeovnV1().Vips().Create(context.Background(), vip, metav1.CreateOptions{}); err != nil {	×
322	klog.Errorf("failed to create health check vip %s, %v", vipName, err)	×
323	return "", err	×
324	}	×
325
326	// wait for vip created
327	time.Sleep(1 * time.Second)	×
328	checkVip, err = c.virtualIpsLister.Get(vipName)	×
329	if err != nil {	×
330	klog.Errorf("failed to get health check vip %s, %v", vipName, err)	×
331	return "", err	×
332	}	×
333	}
334
335	if checkVip.Status.V4ip == "" && checkVip.Status.V6ip == "" {	×
336	err = fmt.Errorf("vip %s is not ready", vipName)	×
337	klog.Error(err)	×
338	return "", err	×
339	}	×
340
341	switch util.CheckProtocol(lbVip) {	×
342	case kubeovnv1.ProtocolIPv4:	×
343	checkIP = checkVip.Status.V4ip	×
344	case kubeovnv1.ProtocolIPv6:	×
345	checkIP = checkVip.Status.V6ip	×
346	}
347	if checkIP == "" {	×
348	err = fmt.Errorf("failed to get health check vip subnet %s", vipName)	×
349	klog.Error(err)	×
350	return "", err	×
351	}	×
352
353	return checkIP, nil	×
354	}
355
356	func getIPPortMappingBackend(endpointSlices []*discoveryv1.EndpointSlice, servicePort v1.ServicePort, serviceIP, checkVip string, isGenIPPortMapping bool) (map[string]string, []string) {	×
357	var (	×
358	ipPortMapping = map[string]string{}	×
359	backends = []string{}	×
360	protocol = util.CheckProtocol(serviceIP)	×
361	)	×
362		×
363	for _, endpointSlice := range endpointSlices {	×
364	var targetPort int32	×
365	for _, port := range endpointSlice.Ports {	×
366	if port.Name != nil && *port.Name == servicePort.Name {	×
367	targetPort = *port.Port	×
368	break	×
369	}
370	}
371	if targetPort == 0 {	×
372	continue	×
373	}
374
375	for _, endpoint := range endpointSlice.Endpoints {	×
376	if isGenIPPortMapping && endpoint.TargetRef.Name != "" {	×
377	lspName := getEndpointTargetLSP(endpoint.TargetRef.Name, endpoint.TargetRef.Namespace, util.OvnProvider)	×
378	for _, address := range endpoint.Addresses {	×
NEW 379	key := address	×
NEW 380	if util.CheckProtocol(address) == kubeovnv1.ProtocolIPv6 {	×
NEW 381	key = fmt.Sprintf("[%s]", address)	×
NEW 382	}	×
NEW 383	ipPortMapping[key] = fmt.Sprintf(util.HealthCheckNamedVipTemplate, lspName, checkVip)	×
384	}
385	}
386	}
387
388	for _, endpoint := range endpointSlice.Endpoints {	×
389	if !endpointReady(endpoint) {	×
390	continue	×
391	}
392
393	for _, address := range endpoint.Addresses {	×
394	if util.CheckProtocol(address) == protocol {	×
395	backends = append(backends, util.JoinHostPort(address, targetPort))	×
396	}	×
397	}
398	}
399	}
400
401	return ipPortMapping, backends	×
402	}
403
404	func endpointReady(endpoint discoveryv1.Endpoint) bool {	2✔
405	return endpoint.Conditions.Ready == nil \|\| *endpoint.Conditions.Ready	2✔
406	}	2✔
407
408	// getEndpointTargetLSP returns the name of the LSP for a given target/namespace.
409	// A custom provider can be specified if the LSP is within a subnet that doesn't use
410	// the default "ovn" provider.
411	func getEndpointTargetLSP(target, namespace, provider string) string {	2✔
412	// This pod seems to be a VM launcher pod, but we do not use the same syntax for the LSP	2✔
413	// of normal pods and for VM pods. We need to retrieve the real name of the VM from	2✔
414	// the pod's name to compute the LSP.	2✔
415	if strings.HasPrefix(target, util.VMLauncherPrefix) {	4✔
416	target = getVMNameFromLauncherPod(target)	2✔
417	}	2✔
418
419	return ovs.PodNameToPortName(target, namespace, provider)	2✔
420	}
421
422	// getVMNameFromLauncherPod returns the name of a VirtualMachine from the name of its launcher pod (virt-launcher)
423	func getVMNameFromLauncherPod(podName string) string {	2✔
424	// Remove the VM launcher pod prefix	2✔
425	vmName := strings.TrimPrefix(podName, util.VMLauncherPrefix)	2✔
426		2✔
427	// Remove the ID of the pod	2✔
428	slice := strings.Split(vmName, "-")	2✔
429	if len(slice) > 0 {	4✔
430	vmName = strings.Join(slice[:len(slice)-1], "-")	2✔
431	}	2✔
432
433	return vmName	2✔
434	}

kubeovn / kube-ovn / 21056235309

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous