16905923923

Committed 12 Aug 2025 10:17AM UTC coverage: 21.406% (-0.006%) from 21.412%

Build # 16905923923

Build Type

push

github

Committed by

web-flow

Commit Message

reduce probability of maps conncurrent iteration and write (#5585)

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

Run Details

0 of 7 new or added lines in 1 file covered. (0.0%)

19 existing lines in 2 files now uncovered.

10579 of 49420 relevant lines covered (21.41%)

0.25 hits per line

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

0.0

/pkg/controller/service.go

package controller

import (
        "context"
        "fmt"
        "maps"
        "net"
        "reflect"
        "slices"
        "strings"
        "time"

        v1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/api/equality"
        k8serrors "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"

        "github.com/kubeovn/kube-ovn/pkg/util"
)

type vpcService struct {
        Vips     []string
        Vpc      string
        Protocol v1.Protocol
        Svc      *v1.Service
}

type updateSvcObject struct {
        key      string
        oldPorts []v1.ServicePort
        newPorts []v1.ServicePort
}

func (c *Controller) enqueueAddService(obj any) {
        svc := obj.(*v1.Service)
        key := cache.MetaObjectToName(svc).String()
        klog.V(3).Infof("enqueue add endpoint %s", key)
        c.addOrUpdateEndpointSliceQueue.Add(key)

        if c.config.EnableNP {
                netpols, err := c.svcMatchNetworkPolicies(svc)
                if err != nil {
                        utilruntime.HandleError(err)
                        return
                }

                for _, np := range netpols {
                        c.updateNpQueue.Add(np)
                }
        }

        if c.config.EnableLbSvc {
                klog.V(3).Infof("enqueue add service %s", key)
                c.addServiceQueue.Add(key)
        }
}

func (c *Controller) enqueueDeleteService(obj any) {
        svc := obj.(*v1.Service)
        klog.Infof("enqueue delete service %s/%s", svc.Namespace, svc.Name)

        vip, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]
        if ok || svc.Spec.ClusterIP != v1.ClusterIPNone && svc.Spec.ClusterIP != "" || svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {
                if c.config.EnableNP {
                        netpols, err := c.svcMatchNetworkPolicies(svc)
                        if err != nil {
                                utilruntime.HandleError(err)
                                return
                        }

                        for _, np := range netpols {
                                c.updateNpQueue.Add(np)
                        }
                }

                ips := util.ServiceClusterIPs(*svc)
                if ok {
                        ips = strings.Split(vip, ",")
                }

                if svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {
                        for _, ingress := range svc.Status.LoadBalancer.Ingress {
                                ips = append(ips, ingress.IP)
                        }
                }

                for _, port := range svc.Spec.Ports {
                        vpcSvc := &vpcService{
                                Protocol: port.Protocol,
                                Vpc:      svc.Annotations[util.VpcAnnotation],
                                Svc:      svc,
                        }
                        for _, ip := range ips {
                                vpcSvc.Vips = append(vpcSvc.Vips, util.JoinHostPort(ip, port.Port))
                        }
                        klog.V(3).Infof("delete vpc service: %v", vpcSvc)
                        c.deleteServiceQueue.Add(vpcSvc)
                }
        }
}

func (c *Controller) enqueueUpdateService(oldObj, newObj any) {
        oldSvc := oldObj.(*v1.Service)
        newSvc := newObj.(*v1.Service)
        if oldSvc.ResourceVersion == newSvc.ResourceVersion {
                return
        }

        oldClusterIps := getVipIps(oldSvc)
        newClusterIps := getVipIps(newSvc)
        var ipsToDel []string
        for _, oldClusterIP := range oldClusterIps {
                if !slices.Contains(newClusterIps, oldClusterIP) {
                        ipsToDel = append(ipsToDel, oldClusterIP)
                }
        }

        key := cache.MetaObjectToName(newSvc).String()
        klog.V(3).Infof("enqueue update service %s", key)
        if len(ipsToDel) != 0 {
                ipsToDelStr := strings.Join(ipsToDel, ",")
                key = strings.Join([]string{key, ipsToDelStr}, "#")
        }

        updateSvc := &updateSvcObject{
                key:      key,
                oldPorts: oldSvc.Spec.Ports,
                newPorts: newSvc.Spec.Ports,
        }
        c.updateServiceQueue.Add(updateSvc)
}

func (c *Controller) handleDeleteService(service *vpcService) error {
        key := cache.MetaObjectToName(service.Svc).String()

        c.svcKeyMutex.LockKey(key)
        defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()
        klog.Infof("handle delete service %s", key)

        svcs, err := c.servicesLister.Services(v1.NamespaceAll).List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list svc, %v", err)
                return err
        }

        var (
                vpcLB             [2]string
                vpcLbConfig       = c.GenVpcLoadBalancer(service.Vpc)
                ignoreHealthCheck = true
        )

        switch service.Protocol {
        case v1.ProtocolTCP:
                vpcLB = [2]string{vpcLbConfig.TCPLoadBalancer, vpcLbConfig.TCPSessLoadBalancer}
        case v1.ProtocolUDP:
                vpcLB = [2]string{vpcLbConfig.UDPLoadBalancer, vpcLbConfig.UDPSessLoadBalancer}
        case v1.ProtocolSCTP:
                vpcLB = [2]string{vpcLbConfig.SctpLoadBalancer, vpcLbConfig.SctpSessLoadBalancer}
        }

        for _, vip := range service.Vips {
                var (
                        ip    string
                        found bool
                )
                ip = parseVipAddr(vip)

                for _, svc := range svcs {
                        if slices.Contains(util.ServiceClusterIPs(*svc), ip) {
                                found = true
                                break
                        }
                }
                if found {
                        continue
                }

                for _, lb := range vpcLB {
                        if err = c.OVNNbClient.LoadBalancerDeleteVip(lb, vip, ignoreHealthCheck); err != nil {
                                klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lb, 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 service.Svc.Spec.Type == v1.ServiceTypeLoadBalancer && c.config.EnableLbSvc {
                if err := c.deleteLbSvc(service.Svc); err != nil {
                        klog.Errorf("failed to delete service %s, %v", service.Svc.Name, err)
                        return err
                }
        }

        return nil
}

func (c *Controller) handleUpdateService(svcObject *updateSvcObject) error {
        key := svcObject.key
        keys := strings.Split(key, "#")
        key = keys[0]
        var ipsToDel []string
        if len(keys) == 2 {
                ipsToDelStr := keys[1]
                ipsToDel = strings.Split(ipsToDelStr, ",")
        }

        namespace, name, err := cache.SplitMetaNamespaceKey(key)
        if err != nil {
                klog.Error(err)
                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
                return nil
        }

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

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

        ips := getVipIps(svc)

        vpcName := svc.Annotations[util.VpcAnnotation]
        if vpcName == "" {
                vpcName = c.config.ClusterRouter
        }
        vpc, err := c.vpcsLister.Get(vpcName)
        if err != nil {
                klog.Errorf("failed to get vpc %s of lb, %v", vpcName, err)
                return err
        }

        tcpLb, udpLb, sctpLb := vpc.Status.TCPLoadBalancer, vpc.Status.UDPLoadBalancer, vpc.Status.SctpLoadBalancer
        oTCPLb, oUDPLb, oSctpLb := vpc.Status.TCPSessionLoadBalancer, vpc.Status.UDPSessionLoadBalancer, vpc.Status.SctpSessionLoadBalancer
        if svc.Spec.SessionAffinity == v1.ServiceAffinityClientIP {
                tcpLb, udpLb, sctpLb, oTCPLb, oUDPLb, oSctpLb = oTCPLb, oUDPLb, oSctpLb, tcpLb, udpLb, sctpLb
        }

        var tcpVips, udpVips, sctpVips []string
        for _, port := range svc.Spec.Ports {
                for _, ip := range ips {
                        switch port.Protocol {
                        case v1.ProtocolTCP:
                                tcpVips = append(tcpVips, util.JoinHostPort(ip, port.Port))
                        case v1.ProtocolUDP:
                                udpVips = append(udpVips, util.JoinHostPort(ip, port.Port))
                        case v1.ProtocolSCTP:
                                sctpVips = append(sctpVips, util.JoinHostPort(ip, port.Port))
                        }
                }
        }

        var (
                needUpdateEndpointQueue = false
                ignoreHealthCheck       = true
        )

        // for service update
        updateVip := func(lbName, oLbName string, svcVips []string) error {
                if len(lbName) == 0 {
                        return nil
                }

                lb, err := c.OVNNbClient.GetLoadBalancer(lbName, false)
                if err != nil {
                        klog.Errorf("failed to get LB %s: %v", lbName, err)
                        return err
                }
                lbVIPs := maps.Clone(lb.Vips)
                klog.V(3).Infof("existing vips of LB %s: %v", lbName, lbVIPs)
                for _, vip := range svcVips {
                        if err := c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {
                                klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)
                                return err
                        }

                        if _, ok := lbVIPs[vip]; !ok {
                                klog.Infof("add vip %s to LB %s", vip, lbName)
                                needUpdateEndpointQueue = true
                        }
                }
                for vip := range lbVIPs {
                        if ip := parseVipAddr(vip); (slices.Contains(ips, ip) && !slices.Contains(svcVips, vip)) || slices.Contains(ipsToDel, ip) {
                                klog.Infof("remove stale vip %s from LB %s", vip, lbName)
                                if err := c.OVNNbClient.LoadBalancerDeleteVip(lbName, vip, ignoreHealthCheck); err != nil {
                                        klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lbName, err)
                                        return err
                                }
                        }
                }

                if len(oLbName) == 0 {
                        return nil
                }

                oLb, err := c.OVNNbClient.GetLoadBalancer(oLbName, false)
                if err != nil {
                        klog.Errorf("failed to get LB %s: %v", oLbName, err)
                        return err
                }
                oLbVIPs := maps.Clone(oLb.Vips)
                klog.V(3).Infof("existing vips of LB %s: %v", oLbName, oLbVIPs)
                for vip := range oLbVIPs {
                        if ip := parseVipAddr(vip); slices.Contains(ips, ip) || slices.Contains(ipsToDel, ip) {
                                klog.Infof("remove stale vip %s from LB %s", vip, oLbName)
                                if err = c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {
                                        klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)
                                        return err
                                }
                        }
                }
                return nil
        }

        if err = updateVip(tcpLb, oTCPLb, tcpVips); err != nil {
                klog.Error(err)
                return err
        }
        if err = updateVip(udpLb, oUDPLb, udpVips); err != nil {
                klog.Error(err)
                return err
        }
        if err = updateVip(sctpLb, oSctpLb, sctpVips); err != nil {
                klog.Error(err)
                return err
        }

        if err := c.checkServiceLBIPBelongToSubnet(svc); err != nil {
                klog.Error(err)
                return err
        }

        if needUpdateEndpointQueue {
                c.addOrUpdateEndpointSliceQueue.Add(key)
        }

        if c.config.EnableLbSvc && svc.Spec.Type == v1.ServiceTypeLoadBalancer {
                changed, err := c.checkLbSvcDeployAnnotationChanged(svc)
                if err != nil {
                        klog.Errorf("failed to check annotation change for lb svc %s: %v", key, err)
                        return err
                }

                // only process svc.spec.ports update
                if !changed {
                        klog.Infof("update loadbalancer service %s", key)
                        pod, err := c.getLbSvcPod(name, namespace)
                        if err != nil {
                                klog.Errorf("failed to get pod for lb svc %s: %v", key, err)
                                if strings.Contains(err.Error(), "not found") {
                                        return nil
                                }
                                return err
                        }

                        toDel := diffSvcPorts(svcObject.oldPorts, svcObject.newPorts)
                        if err := c.delDnatRules(pod, toDel, svc); err != nil {
                                klog.Errorf("failed to delete dnat rules, err: %v", err)
                                return err
                        }
                        if err = c.updatePodAttachNets(pod, svc); err != nil {
                                klog.Errorf("failed to update pod attachment network for lb svc %s: %v", key, err)
                                return err
                        }
                }
        }

        return nil
}

// Parse key of map, [fd00:10:96::11c9]:10665 for example
func parseVipAddr(vip string) string {
        host, _, err := net.SplitHostPort(vip)
        if err != nil {
                klog.Errorf("failed to parse vip %q: %v", vip, err)
                return ""
        }
        return host
}

func (c *Controller) handleAddService(key string) error {
        if !c.config.EnableLbSvc {
                return nil
        }

        namespace, name, err := cache.SplitMetaNamespaceKey(key)
        if err != nil {
                klog.Error(err)
                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
                return nil
        }

        c.svcKeyMutex.LockKey(key)
        defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()
        klog.Infof("handle add service %s", key)

        svc, err := c.servicesLister.Services(namespace).Get(name)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        if svc.Spec.Type != v1.ServiceTypeLoadBalancer {
                return nil
        }
        // Skip non kube-ovn lb-svc.
        if _, ok := svc.Annotations[util.AttachmentProvider]; !ok {
                return nil
        }

        klog.Infof("handle add loadbalancer service %s", key)

        if err = c.validateSvc(svc); err != nil {
                c.recorder.Event(svc, v1.EventTypeWarning, "ValidateSvcFailed", err.Error())
                klog.Errorf("failed to validate lb svc %s: %v", key, err)
                return err
        }

        nad, err := c.getAttachNetworkForService(svc)
        if err != nil {
                c.recorder.Event(svc, v1.EventTypeWarning, "GetNADFailed", err.Error())
                klog.Errorf("failed to check attachment network of lb svc %s: %v", key, err)
                return err
        }

        if err = c.createLbSvcPod(svc, nad); err != nil {
                klog.Errorf("failed to create lb svc pod for %s: %v", key, err)
                return err
        }

        var pod *v1.Pod
        for {
                pod, err = c.getLbSvcPod(name, namespace)
                if err != nil {
                        klog.Warningf("pod for lb svc %s is not running: %v", key, err)
                        time.Sleep(time.Second)
                }
                if pod != nil {
                        break
                }

                // It's important here to check existing of svc, used to break the loop.
                _, err = c.servicesLister.Services(namespace).Get(name)
                if err != nil {
                        if k8serrors.IsNotFound(err) {
                                return nil
                        }
                        klog.Error(err)
                        return err
                }
        }

        loadBalancerIP, err := c.getPodAttachIP(pod, svc)
        if err != nil {
                klog.Errorf("failed to get loadBalancerIP: %v", err)
                return err
        }

        svc, err = c.servicesLister.Services(namespace).Get(name)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        targetSvc := svc.DeepCopy()
        if err = c.updatePodAttachNets(pod, targetSvc); err != nil {
                klog.Errorf("failed to update pod attachment network for service %s/%s: %v", namespace, name, err)
                return err
        }

        // compatible with IPv4 and IPv6 dual stack subnet.
        var ingress []v1.LoadBalancerIngress
        for ip := range strings.SplitSeq(loadBalancerIP, ",") {
                if ip != "" && net.ParseIP(ip) != nil {
                        ingress = append(ingress, v1.LoadBalancerIngress{IP: ip})
                }
        }
        targetSvc.Status.LoadBalancer.Ingress = ingress
        if !equality.Semantic.DeepEqual(svc.Status, targetSvc.Status) {
                if _, err = c.config.KubeClient.CoreV1().Services(namespace).
                        UpdateStatus(context.Background(), targetSvc, metav1.UpdateOptions{}); err != nil {
                        klog.Errorf("failed to update status of service %s/%s: %v", namespace, name, err)
                        return err
                }
        }

        return nil
}

func getVipIps(svc *v1.Service) []string {
        var ips []string
        if vip, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {
                ips = strings.Split(vip, ",")
        } else {
                ips = util.ServiceClusterIPs(*svc)
                if svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {
                        for _, ingress := range svc.Status.LoadBalancer.Ingress {
                                ips = append(ips, ingress.IP)
                        }
                }
        }
        return ips
}

func diffSvcPorts(oldPorts, newPorts []v1.ServicePort) (toDel []v1.ServicePort) {
        for _, oldPort := range oldPorts {
                found := false
                for _, newPort := range newPorts {
                        if reflect.DeepEqual(oldPort, newPort) {
                                found = true
                                break
                        }
                }
                if !found {
                        toDel = append(toDel, oldPort)
                }
        }

        return toDel
}

func (c *Controller) checkServiceLBIPBelongToSubnet(svc *v1.Service) error {
        subnets, err := c.subnetsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list subnets: %v", err)
                return err
        }

        isServiceExternalIPFromSubnet := false
        for _, subnet := range subnets {
                for _, ingress := range svc.Status.LoadBalancer.Ingress {
                        if util.CIDRContainIP(subnet.Spec.CIDRBlock, ingress.IP) {
                                svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] = subnet.Name
                                isServiceExternalIPFromSubnet = true
                                break
                        }
                }
        }

        if !isServiceExternalIPFromSubnet {
                delete(svc.Annotations, util.ServiceExternalIPFromSubnetAnnotation)
        }
        klog.Infof("Service %s/%s external IP belongs to subnet: %v", svc.Namespace, svc.Name, isServiceExternalIPFromSubnet)
        if _, err = c.config.KubeClient.CoreV1().Services(svc.Namespace).Update(context.TODO(), svc, metav1.UpdateOptions{}); err != nil {
                klog.Errorf("failed to update service %s/%s: %v", svc.Namespace, svc.Name, err)
                return err
        }

        return nil
}

1	package controller
2
3	import (
4	"context"
5	"fmt"
6	"maps"
7	"net"
8	"reflect"
9	"slices"
10	"strings"
11	"time"
12
13	v1 "k8s.io/api/core/v1"
14	"k8s.io/apimachinery/pkg/api/equality"
15	k8serrors "k8s.io/apimachinery/pkg/api/errors"
16	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
17	"k8s.io/apimachinery/pkg/labels"
18	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
19	"k8s.io/client-go/tools/cache"
20	"k8s.io/klog/v2"
21
22	"github.com/kubeovn/kube-ovn/pkg/util"
23	)
24
25	type vpcService struct {
26	Vips []string
27	Vpc string
28	Protocol v1.Protocol
29	Svc *v1.Service
30	}
31
32	type updateSvcObject struct {
33	key string
34	oldPorts []v1.ServicePort
35	newPorts []v1.ServicePort
36	}
37
38	func (c *Controller) enqueueAddService(obj any) {	×
39	svc := obj.(*v1.Service)	×
40	key := cache.MetaObjectToName(svc).String()	×
41	klog.V(3).Infof("enqueue add endpoint %s", key)	×
42	c.addOrUpdateEndpointSliceQueue.Add(key)	×
43		×
44	if c.config.EnableNP {	×
45	netpols, err := c.svcMatchNetworkPolicies(svc)	×
46	if err != nil {	×
47	utilruntime.HandleError(err)	×
48	return	×
49	}	×
50
51	for _, np := range netpols {	×
52	c.updateNpQueue.Add(np)	×
53	}	×
54	}
55
56	if c.config.EnableLbSvc {	×
57	klog.V(3).Infof("enqueue add service %s", key)	×
58	c.addServiceQueue.Add(key)	×
59	}	×
60	}
61
62	func (c *Controller) enqueueDeleteService(obj any) {	×
63	svc := obj.(*v1.Service)	×
64	klog.Infof("enqueue delete service %s/%s", svc.Namespace, svc.Name)	×
65		×
66	vip, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]	×
67	if ok \|\| svc.Spec.ClusterIP != v1.ClusterIPNone && svc.Spec.ClusterIP != "" \|\| svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {	×
68	if c.config.EnableNP {	×
69	netpols, err := c.svcMatchNetworkPolicies(svc)	×
70	if err != nil {	×
71	utilruntime.HandleError(err)	×
72	return	×
73	}	×
74
75	for _, np := range netpols {	×
76	c.updateNpQueue.Add(np)	×
77	}	×
78	}
79
80	ips := util.ServiceClusterIPs(*svc)	×
81	if ok {	×
82	ips = strings.Split(vip, ",")	×
83	}	×
84
85	if svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {	×
86	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
87	ips = append(ips, ingress.IP)	×
88	}	×
89	}
90
91	for _, port := range svc.Spec.Ports {	×
92	vpcSvc := &vpcService{	×
93	Protocol: port.Protocol,	×
94	Vpc: svc.Annotations[util.VpcAnnotation],	×
95	Svc: svc,	×
96	}	×
97	for _, ip := range ips {	×
98	vpcSvc.Vips = append(vpcSvc.Vips, util.JoinHostPort(ip, port.Port))	×
99	}	×
100	klog.V(3).Infof("delete vpc service: %v", vpcSvc)	×
101	c.deleteServiceQueue.Add(vpcSvc)	×
102	}
103	}
104	}
105
106	func (c *Controller) enqueueUpdateService(oldObj, newObj any) {	×
107	oldSvc := oldObj.(*v1.Service)	×
108	newSvc := newObj.(*v1.Service)	×
109	if oldSvc.ResourceVersion == newSvc.ResourceVersion {	×
110	return	×
111	}	×
112
113	oldClusterIps := getVipIps(oldSvc)	×
114	newClusterIps := getVipIps(newSvc)	×
115	var ipsToDel []string	×
116	for _, oldClusterIP := range oldClusterIps {	×
117	if !slices.Contains(newClusterIps, oldClusterIP) {	×
118	ipsToDel = append(ipsToDel, oldClusterIP)	×
119	}	×
120	}
121
122	key := cache.MetaObjectToName(newSvc).String()	×
123	klog.V(3).Infof("enqueue update service %s", key)	×
124	if len(ipsToDel) != 0 {	×
125	ipsToDelStr := strings.Join(ipsToDel, ",")	×
126	key = strings.Join([]string{key, ipsToDelStr}, "#")	×
127	}	×
128
129	updateSvc := &updateSvcObject{	×
130	key: key,	×
131	oldPorts: oldSvc.Spec.Ports,	×
132	newPorts: newSvc.Spec.Ports,	×
133	}	×
134	c.updateServiceQueue.Add(updateSvc)	×
135	}
136
137	func (c Controller) handleDeleteService(service vpcService) error {	×
138	key := cache.MetaObjectToName(service.Svc).String()	×
139		×
140	c.svcKeyMutex.LockKey(key)	×
141	defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()	×
142	klog.Infof("handle delete service %s", key)	×
143		×
144	svcs, err := c.servicesLister.Services(v1.NamespaceAll).List(labels.Everything())	×
145	if err != nil {	×
146	klog.Errorf("failed to list svc, %v", err)	×
147	return err	×
148	}	×
149
150	var (	×
151	vpcLB [2]string	×
152	vpcLbConfig = c.GenVpcLoadBalancer(service.Vpc)	×
153	ignoreHealthCheck = true	×
154	)	×
155		×
156	switch service.Protocol {	×
157	case v1.ProtocolTCP:	×
158	vpcLB = [2]string{vpcLbConfig.TCPLoadBalancer, vpcLbConfig.TCPSessLoadBalancer}	×
159	case v1.ProtocolUDP:	×
160	vpcLB = [2]string{vpcLbConfig.UDPLoadBalancer, vpcLbConfig.UDPSessLoadBalancer}	×
161	case v1.ProtocolSCTP:	×
162	vpcLB = [2]string{vpcLbConfig.SctpLoadBalancer, vpcLbConfig.SctpSessLoadBalancer}	×
163	}
164
165	for _, vip := range service.Vips {	×
166	var (	×
167	ip string	×
168	found bool	×
169	)	×
170	ip = parseVipAddr(vip)	×
171		×
172	for _, svc := range svcs {	×
173	if slices.Contains(util.ServiceClusterIPs(*svc), ip) {	×
174	found = true	×
175	break	×
176	}
177	}
178	if found {	×
179	continue	×
180	}
181
182	for _, lb := range vpcLB {	×
183	if err = c.OVNNbClient.LoadBalancerDeleteVip(lb, vip, ignoreHealthCheck); err != nil {	×
184	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lb, err)	×
185	return err	×
186	}	×
187
188	if c.config.EnableOVNLBPreferLocal {	×
189	if err = c.OVNNbClient.LoadBalancerDeleteIPPortMapping(lb, vip); err != nil {	×
190	klog.Errorf("failed to delete ip port mapping for vip %s from LB %s: %v", vip, lb, err)	×
191	return err	×
192	}	×
193	}
194	}
195	}
196
197	if service.Svc.Spec.Type == v1.ServiceTypeLoadBalancer && c.config.EnableLbSvc {	×
198	if err := c.deleteLbSvc(service.Svc); err != nil {	×
199	klog.Errorf("failed to delete service %s, %v", service.Svc.Name, err)	×
200	return err	×
201	}	×
202	}
203
204	return nil	×
205	}
206
207	func (c Controller) handleUpdateService(svcObject updateSvcObject) error {	×
208	key := svcObject.key	×
209	keys := strings.Split(key, "#")	×
210	key = keys[0]	×
211	var ipsToDel []string	×
212	if len(keys) == 2 {	×
213	ipsToDelStr := keys[1]	×
214	ipsToDel = strings.Split(ipsToDelStr, ",")	×
215	}	×
216
217	namespace, name, err := cache.SplitMetaNamespaceKey(key)	×
218	if err != nil {	×
219	klog.Error(err)	×
220	utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))	×
221	return nil	×
222	}	×
223
224	c.svcKeyMutex.LockKey(key)	×
225	defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()	×
226	klog.Infof("handle update service %s", key)	×
227		×
228	svc, err := c.servicesLister.Services(namespace).Get(name)	×
229	if err != nil {	×
230	if k8serrors.IsNotFound(err) {	×
231	return nil	×
232	}	×
233	klog.Error(err)	×
234	return err	×
235	}
236
237	ips := getVipIps(svc)	×
238		×
239	vpcName := svc.Annotations[util.VpcAnnotation]	×
240	if vpcName == "" {	×
241	vpcName = c.config.ClusterRouter	×
242	}	×
243	vpc, err := c.vpcsLister.Get(vpcName)	×
244	if err != nil {	×
245	klog.Errorf("failed to get vpc %s of lb, %v", vpcName, err)	×
246	return err	×
247	}	×
248
249	tcpLb, udpLb, sctpLb := vpc.Status.TCPLoadBalancer, vpc.Status.UDPLoadBalancer, vpc.Status.SctpLoadBalancer	×
250	oTCPLb, oUDPLb, oSctpLb := vpc.Status.TCPSessionLoadBalancer, vpc.Status.UDPSessionLoadBalancer, vpc.Status.SctpSessionLoadBalancer	×
251	if svc.Spec.SessionAffinity == v1.ServiceAffinityClientIP {	×
252	tcpLb, udpLb, sctpLb, oTCPLb, oUDPLb, oSctpLb = oTCPLb, oUDPLb, oSctpLb, tcpLb, udpLb, sctpLb	×
253	}	×
254
255	var tcpVips, udpVips, sctpVips []string	×
256	for _, port := range svc.Spec.Ports {	×
257	for _, ip := range ips {	×
258	switch port.Protocol {	×
259	case v1.ProtocolTCP:	×
260	tcpVips = append(tcpVips, util.JoinHostPort(ip, port.Port))	×
261	case v1.ProtocolUDP:	×
262	udpVips = append(udpVips, util.JoinHostPort(ip, port.Port))	×
263	case v1.ProtocolSCTP:	×
264	sctpVips = append(sctpVips, util.JoinHostPort(ip, port.Port))	×
265	}
266	}
267	}
268
269	var (	×
270	needUpdateEndpointQueue = false	×
271	ignoreHealthCheck = true	×
272	)	×
273		×
274	// for service update	×
275	updateVip := func(lbName, oLbName string, svcVips []string) error {	×
276	if len(lbName) == 0 {	×
277	return nil	×
278	}	×
279
280	lb, err := c.OVNNbClient.GetLoadBalancer(lbName, false)	×
281	if err != nil {	×
282	klog.Errorf("failed to get LB %s: %v", lbName, err)	×
283	return err	×
284	}	×
NEW 285	lbVIPs := maps.Clone(lb.Vips)	×
NEW 286	klog.V(3).Infof("existing vips of LB %s: %v", lbName, lbVIPs)	×
287	for _, vip := range svcVips {	×
288	if err := c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {	×
289	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)	×
290	return err	×
291	}	×
292
NEW 293	if _, ok := lbVIPs[vip]; !ok {	×
294	klog.Infof("add vip %s to LB %s", vip, lbName)	×
295	needUpdateEndpointQueue = true	×
296	}	×
297	}
NEW 298	for vip := range lbVIPs {	×
299	if ip := parseVipAddr(vip); (slices.Contains(ips, ip) && !slices.Contains(svcVips, vip)) \|\| slices.Contains(ipsToDel, ip) {	×
300	klog.Infof("remove stale vip %s from LB %s", vip, lbName)	×
301	if err := c.OVNNbClient.LoadBalancerDeleteVip(lbName, vip, ignoreHealthCheck); err != nil {	×
302	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lbName, err)	×
303	return err	×
304	}	×
305	}
306	}
307
308	if len(oLbName) == 0 {	×
309	return nil	×
310	}	×
311
312	oLb, err := c.OVNNbClient.GetLoadBalancer(oLbName, false)	×
313	if err != nil {	×
314	klog.Errorf("failed to get LB %s: %v", oLbName, err)	×
315	return err	×
316	}	×
NEW 317	oLbVIPs := maps.Clone(oLb.Vips)	×
NEW 318	klog.V(3).Infof("existing vips of LB %s: %v", oLbName, oLbVIPs)	×
NEW 319	for vip := range oLbVIPs {	×
320	if ip := parseVipAddr(vip); slices.Contains(ips, ip) \|\| slices.Contains(ipsToDel, ip) {	×
321	klog.Infof("remove stale vip %s from LB %s", vip, oLbName)	×
322	if err = c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {	×
323	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)	×
324	return err	×
325	}	×
326	}
327	}
328	return nil	×
329	}
330
331	if err = updateVip(tcpLb, oTCPLb, tcpVips); err != nil {	×
332	klog.Error(err)	×
333	return err	×
334	}	×
335	if err = updateVip(udpLb, oUDPLb, udpVips); err != nil {	×
336	klog.Error(err)	×
337	return err	×
338	}	×
339	if err = updateVip(sctpLb, oSctpLb, sctpVips); err != nil {	×
340	klog.Error(err)	×
341	return err	×
342	}	×
343
344	if err := c.checkServiceLBIPBelongToSubnet(svc); err != nil {	×
345	klog.Error(err)	×
346	return err	×
347	}	×
348
349	if needUpdateEndpointQueue {	×
350	c.addOrUpdateEndpointSliceQueue.Add(key)	×
351	}	×
352
353	if c.config.EnableLbSvc && svc.Spec.Type == v1.ServiceTypeLoadBalancer {	×
354	changed, err := c.checkLbSvcDeployAnnotationChanged(svc)	×
355	if err != nil {	×
356	klog.Errorf("failed to check annotation change for lb svc %s: %v", key, err)	×
357	return err	×
358	}	×
359
360	// only process svc.spec.ports update
361	if !changed {	×
362	klog.Infof("update loadbalancer service %s", key)	×
363	pod, err := c.getLbSvcPod(name, namespace)	×
364	if err != nil {	×
365	klog.Errorf("failed to get pod for lb svc %s: %v", key, err)	×
366	if strings.Contains(err.Error(), "not found") {	×
367	return nil	×
368	}	×
369	return err	×
370	}
371
372	toDel := diffSvcPorts(svcObject.oldPorts, svcObject.newPorts)	×
373	if err := c.delDnatRules(pod, toDel, svc); err != nil {	×
374	klog.Errorf("failed to delete dnat rules, err: %v", err)	×
375	return err	×
376	}	×
377	if err = c.updatePodAttachNets(pod, svc); err != nil {	×
378	klog.Errorf("failed to update pod attachment network for lb svc %s: %v", key, err)	×
379	return err	×
380	}	×
381	}
382	}
383
384	return nil	×
385	}
386
387	// Parse key of map, [fd00:10:96::11c9]:10665 for example
388	func parseVipAddr(vip string) string {	×
389	host, _, err := net.SplitHostPort(vip)	×
390	if err != nil {	×
391	klog.Errorf("failed to parse vip %q: %v", vip, err)	×
392	return ""	×
393	}	×
394	return host	×
395	}
396
397	func (c *Controller) handleAddService(key string) error {	×
398	if !c.config.EnableLbSvc {	×
399	return nil	×
400	}	×
401
402	namespace, name, err := cache.SplitMetaNamespaceKey(key)	×
403	if err != nil {	×
404	klog.Error(err)	×
405	utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))	×
406	return nil	×
407	}	×
408
409	c.svcKeyMutex.LockKey(key)	×
410	defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()	×
411	klog.Infof("handle add service %s", key)	×
412		×
413	svc, err := c.servicesLister.Services(namespace).Get(name)	×
414	if err != nil {	×
415	if k8serrors.IsNotFound(err) {	×
416	return nil	×
417	}	×
418	klog.Error(err)	×
419	return err	×
420	}
421	if svc.Spec.Type != v1.ServiceTypeLoadBalancer {	×
422	return nil	×
423	}	×
424	// Skip non kube-ovn lb-svc.
425	if _, ok := svc.Annotations[util.AttachmentProvider]; !ok {	×
426	return nil	×
427	}	×
428
429	klog.Infof("handle add loadbalancer service %s", key)	×
430		×
431	if err = c.validateSvc(svc); err != nil {	×
432	c.recorder.Event(svc, v1.EventTypeWarning, "ValidateSvcFailed", err.Error())	×
433	klog.Errorf("failed to validate lb svc %s: %v", key, err)	×
434	return err	×
435	}	×
436
437	nad, err := c.getAttachNetworkForService(svc)	×
438	if err != nil {	×
439	c.recorder.Event(svc, v1.EventTypeWarning, "GetNADFailed", err.Error())	×
440	klog.Errorf("failed to check attachment network of lb svc %s: %v", key, err)	×
441	return err	×
442	}	×
443
444	if err = c.createLbSvcPod(svc, nad); err != nil {	×
445	klog.Errorf("failed to create lb svc pod for %s: %v", key, err)	×
446	return err	×
447	}	×
448
449	var pod *v1.Pod	×
450	for {	×
451	pod, err = c.getLbSvcPod(name, namespace)	×
452	if err != nil {	×
453	klog.Warningf("pod for lb svc %s is not running: %v", key, err)	×
454	time.Sleep(time.Second)	×
455	}	×
456	if pod != nil {	×
457	break	×
458	}
459
460	// It's important here to check existing of svc, used to break the loop.
461	_, err = c.servicesLister.Services(namespace).Get(name)	×
462	if err != nil {	×
463	if k8serrors.IsNotFound(err) {	×
464	return nil	×
465	}	×
466	klog.Error(err)	×
467	return err	×
468	}
469	}
470
471	loadBalancerIP, err := c.getPodAttachIP(pod, svc)	×
472	if err != nil {	×
473	klog.Errorf("failed to get loadBalancerIP: %v", err)	×
474	return err	×
475	}	×
476
477	svc, err = c.servicesLister.Services(namespace).Get(name)	×
478	if err != nil {	×
479	if k8serrors.IsNotFound(err) {	×
480	return nil	×
481	}	×
482	klog.Error(err)	×
483	return err	×
484	}
485	targetSvc := svc.DeepCopy()	×
486	if err = c.updatePodAttachNets(pod, targetSvc); err != nil {	×
487	klog.Errorf("failed to update pod attachment network for service %s/%s: %v", namespace, name, err)	×
488	return err	×
489	}	×
490
491	// compatible with IPv4 and IPv6 dual stack subnet.
492	var ingress []v1.LoadBalancerIngress	×
493	for ip := range strings.SplitSeq(loadBalancerIP, ",") {	×
494	if ip != "" && net.ParseIP(ip) != nil {	×
495	ingress = append(ingress, v1.LoadBalancerIngress{IP: ip})	×
496	}	×
497	}
498	targetSvc.Status.LoadBalancer.Ingress = ingress	×
499	if !equality.Semantic.DeepEqual(svc.Status, targetSvc.Status) {	×
500	if _, err = c.config.KubeClient.CoreV1().Services(namespace).	×
501	UpdateStatus(context.Background(), targetSvc, metav1.UpdateOptions{}); err != nil {	×
502	klog.Errorf("failed to update status of service %s/%s: %v", namespace, name, err)	×
503	return err	×
504	}	×
505	}
506
507	return nil	×
508	}
509
510	func getVipIps(svc *v1.Service) []string {	×
511	var ips []string	×
512	if vip, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {	×
513	ips = strings.Split(vip, ",")	×
514	} else {	×
515	ips = util.ServiceClusterIPs(*svc)	×
516	if svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {	×
517	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
518	ips = append(ips, ingress.IP)	×
519	}	×
520	}
521	}
522	return ips	×
523	}
524
525	func diffSvcPorts(oldPorts, newPorts []v1.ServicePort) (toDel []v1.ServicePort) {	×
526	for _, oldPort := range oldPorts {	×
527	found := false	×
528	for _, newPort := range newPorts {	×
529	if reflect.DeepEqual(oldPort, newPort) {	×
530	found = true	×
531	break	×
532	}
533	}
534	if !found {	×
535	toDel = append(toDel, oldPort)	×
536	}	×
537	}
538
539	return toDel	×
540	}
541
542	func (c Controller) checkServiceLBIPBelongToSubnet(svc v1.Service) error {	×
543	subnets, err := c.subnetsLister.List(labels.Everything())	×
544	if err != nil {	×
545	klog.Errorf("failed to list subnets: %v", err)	×
546	return err	×
547	}	×
548
549	isServiceExternalIPFromSubnet := false	×
550	for _, subnet := range subnets {	×
551	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
552	if util.CIDRContainIP(subnet.Spec.CIDRBlock, ingress.IP) {	×
553	svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] = subnet.Name	×
554	isServiceExternalIPFromSubnet = true	×
555	break	×
556	}
557	}
558	}
559
560	if !isServiceExternalIPFromSubnet {	×
561	delete(svc.Annotations, util.ServiceExternalIPFromSubnetAnnotation)	×
562	}	×
563	klog.Infof("Service %s/%s external IP belongs to subnet: %v", svc.Namespace, svc.Name, isServiceExternalIPFromSubnet)	×
564	if _, err = c.config.KubeClient.CoreV1().Services(svc.Namespace).Update(context.TODO(), svc, metav1.UpdateOptions{}); err != nil {	×
565	klog.Errorf("failed to update service %s/%s: %v", svc.Namespace, svc.Name, err)	×
566	return err	×
567	}	×
568
569	return nil	×
570	}

kubeovn / kube-ovn / 16905923923

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