15018874925

Committed 14 May 2025 10:54AM UTC coverage: 21.756% (+0.03%) from 21.723%

Build # 15018874925

Build Type

push

github

Committed by

oilbeater

Commit Message

fix slbr missing vip when svc is deleted

Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>

Coverage Stats

0 of 5 new or added lines in 2 files covered. (0.0%)

86 existing lines in 1 file now uncovered.

10265 of 47182 relevant lines covered (21.76%)

0.25 hits per line

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

2.71

/pkg/controller/endpoint_slice.go

package controller

import (
        "context"
        "fmt"
        "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/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 {
                                        if addr == pod.Status.PodIP {
                                                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 != "" {
                                ipName := fmt.Sprintf("%s.%s", endpoint.TargetRef.Name, endpoint.TargetRef.Namespace)
                                for _, address := range endpoint.Addresses {
                                        ipPortMapping[address] = fmt.Sprintf(util.HealthCheckNamedVipTemplate, ipName, 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
}

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

kubeovn / kube-ovn / 15018874925

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