16863016720

Committed 10 Aug 2025 03:04PM UTC coverage: 21.444% (-0.005%) from 21.449%

Build # 16863016720

Build Type

Pull #5584

github

Committed by

oilbeater

Commit Message

controller: avoid concurrent map access in service update

Pull Request Pull Request #5584: controller: avoid concurrent map access in service update

Run Details

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

2 existing lines in 1 file now uncovered.

10569 of 49287 relevant lines covered (21.44%)

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	}	×
285
NEW 286	lbVips := maps.Clone(lb.Vips)	×
NEW 287	klog.V(3).Infof("existing vips of LB %s: %v", lbName, lbVips)	×
288	for _, vip := range svcVips {	×
289	if err := c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {	×
290	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)	×
291	return err	×
292	}	×
293
NEW 294	if _, ok := lbVips[vip]; !ok {	×
295	klog.Infof("add vip %s to LB %s", vip, lbName)	×
296	needUpdateEndpointQueue = true	×
297	}	×
298	}
299
NEW 300	for vip := range lbVips {	×
301	if ip := parseVipAddr(vip); (slices.Contains(ips, ip) && !slices.Contains(svcVips, vip)) \|\| slices.Contains(ipsToDel, ip) {	×
302	klog.Infof("remove stale vip %s from LB %s", vip, lbName)	×
303	if err := c.OVNNbClient.LoadBalancerDeleteVip(lbName, vip, ignoreHealthCheck); err != nil {	×
304	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lbName, err)	×
305	return err	×
306	}	×
307	}
308	}
309
310	if len(oLbName) == 0 {	×
311	return nil	×
312	}	×
313
314	oLb, err := c.OVNNbClient.GetLoadBalancer(oLbName, false)	×
315	if err != nil {	×
316	klog.Errorf("failed to get LB %s: %v", oLbName, err)	×
317	return err	×
318	}	×
319
NEW 320	oLbVips := maps.Clone(oLb.Vips)	×
NEW 321	klog.V(3).Infof("existing vips of LB %s: %v", oLbName, oLbVips)	×
NEW 322	for vip := range oLbVips {	×
323	if ip := parseVipAddr(vip); slices.Contains(ips, ip) \|\| slices.Contains(ipsToDel, ip) {	×
324	klog.Infof("remove stale vip %s from LB %s", vip, oLbName)	×
325	if err = c.OVNNbClient.LoadBalancerDeleteVip(oLbName, vip, ignoreHealthCheck); err != nil {	×
326	klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oLbName, err)	×
327	return err	×
328	}	×
329	}
330	}
331	return nil	×
332	}
333
334	if err = updateVip(tcpLb, oTCPLb, tcpVips); err != nil {	×
335	klog.Error(err)	×
336	return err	×
337	}	×
338	if err = updateVip(udpLb, oUDPLb, udpVips); err != nil {	×
339	klog.Error(err)	×
340	return err	×
341	}	×
342	if err = updateVip(sctpLb, oSctpLb, sctpVips); err != nil {	×
343	klog.Error(err)	×
344	return err	×
345	}	×
346
347	if err := c.checkServiceLBIPBelongToSubnet(svc); err != nil {	×
348	klog.Error(err)	×
349	return err	×
350	}	×
351
352	if needUpdateEndpointQueue {	×
353	c.addOrUpdateEndpointSliceQueue.Add(key)	×
354	}	×
355
356	if c.config.EnableLbSvc && svc.Spec.Type == v1.ServiceTypeLoadBalancer {	×
357	changed, err := c.checkLbSvcDeployAnnotationChanged(svc)	×
358	if err != nil {	×
359	klog.Errorf("failed to check annotation change for lb svc %s: %v", key, err)	×
360	return err	×
361	}	×
362
363	// only process svc.spec.ports update
364	if !changed {	×
365	klog.Infof("update loadbalancer service %s", key)	×
366	pod, err := c.getLbSvcPod(name, namespace)	×
367	if err != nil {	×
368	klog.Errorf("failed to get pod for lb svc %s: %v", key, err)	×
369	if strings.Contains(err.Error(), "not found") {	×
370	return nil	×
371	}	×
372	return err	×
373	}
374
375	toDel := diffSvcPorts(svcObject.oldPorts, svcObject.newPorts)	×
376	if err := c.delDnatRules(pod, toDel, svc); err != nil {	×
377	klog.Errorf("failed to delete dnat rules, err: %v", err)	×
378	return err	×
379	}	×
380	if err = c.updatePodAttachNets(pod, svc); err != nil {	×
381	klog.Errorf("failed to update pod attachment network for lb svc %s: %v", key, err)	×
382	return err	×
383	}	×
384	}
385	}
386
387	return nil	×
388	}
389
390	// Parse key of map, [fd00:10:96::11c9]:10665 for example
391	func parseVipAddr(vip string) string {	×
392	host, _, err := net.SplitHostPort(vip)	×
393	if err != nil {	×
394	klog.Errorf("failed to parse vip %q: %v", vip, err)	×
395	return ""	×
396	}	×
397	return host	×
398	}
399
400	func (c *Controller) handleAddService(key string) error {	×
401	if !c.config.EnableLbSvc {	×
402	return nil	×
403	}	×
404
405	namespace, name, err := cache.SplitMetaNamespaceKey(key)	×
406	if err != nil {	×
407	klog.Error(err)	×
408	utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))	×
409	return nil	×
410	}	×
411
412	c.svcKeyMutex.LockKey(key)	×
413	defer func() { _ = c.svcKeyMutex.UnlockKey(key) }()	×
414	klog.Infof("handle add service %s", key)	×
415		×
416	svc, err := c.servicesLister.Services(namespace).Get(name)	×
417	if err != nil {	×
418	if k8serrors.IsNotFound(err) {	×
419	return nil	×
420	}	×
421	klog.Error(err)	×
422	return err	×
423	}
424	if svc.Spec.Type != v1.ServiceTypeLoadBalancer {	×
425	return nil	×
426	}	×
427	// Skip non kube-ovn lb-svc.
428	if _, ok := svc.Annotations[util.AttachmentProvider]; !ok {	×
429	return nil	×
430	}	×
431
432	klog.Infof("handle add loadbalancer service %s", key)	×
433		×
434	if err = c.validateSvc(svc); err != nil {	×
435	c.recorder.Event(svc, v1.EventTypeWarning, "ValidateSvcFailed", err.Error())	×
436	klog.Errorf("failed to validate lb svc %s: %v", key, err)	×
437	return err	×
438	}	×
439
440	nad, err := c.getAttachNetworkForService(svc)	×
441	if err != nil {	×
442	c.recorder.Event(svc, v1.EventTypeWarning, "GetNADFailed", err.Error())	×
443	klog.Errorf("failed to check attachment network of lb svc %s: %v", key, err)	×
444	return err	×
445	}	×
446
447	if err = c.createLbSvcPod(svc, nad); err != nil {	×
448	klog.Errorf("failed to create lb svc pod for %s: %v", key, err)	×
449	return err	×
450	}	×
451
452	var pod *v1.Pod	×
453	for {	×
454	pod, err = c.getLbSvcPod(name, namespace)	×
455	if err != nil {	×
456	klog.Warningf("pod for lb svc %s is not running: %v", key, err)	×
457	time.Sleep(time.Second)	×
458	}	×
459	if pod != nil {	×
460	break	×
461	}
462
463	// It's important here to check existing of svc, used to break the loop.
464	_, err = c.servicesLister.Services(namespace).Get(name)	×
465	if err != nil {	×
466	if k8serrors.IsNotFound(err) {	×
467	return nil	×
468	}	×
469	klog.Error(err)	×
470	return err	×
471	}
472	}
473
474	loadBalancerIP, err := c.getPodAttachIP(pod, svc)	×
475	if err != nil {	×
476	klog.Errorf("failed to get loadBalancerIP: %v", err)	×
477	return err	×
478	}	×
479
480	svc, err = c.servicesLister.Services(namespace).Get(name)	×
481	if err != nil {	×
482	if k8serrors.IsNotFound(err) {	×
483	return nil	×
484	}	×
485	klog.Error(err)	×
486	return err	×
487	}
488	targetSvc := svc.DeepCopy()	×
489	if err = c.updatePodAttachNets(pod, targetSvc); err != nil {	×
490	klog.Errorf("failed to update pod attachment network for service %s/%s: %v", namespace, name, err)	×
491	return err	×
492	}	×
493
494	// compatible with IPv4 and IPv6 dual stack subnet.
495	var ingress []v1.LoadBalancerIngress	×
496	for ip := range strings.SplitSeq(loadBalancerIP, ",") {	×
497	if ip != "" && net.ParseIP(ip) != nil {	×
498	ingress = append(ingress, v1.LoadBalancerIngress{IP: ip})	×
499	}	×
500	}
501	targetSvc.Status.LoadBalancer.Ingress = ingress	×
502	if !equality.Semantic.DeepEqual(svc.Status, targetSvc.Status) {	×
503	if _, err = c.config.KubeClient.CoreV1().Services(namespace).	×
504	UpdateStatus(context.Background(), targetSvc, metav1.UpdateOptions{}); err != nil {	×
505	klog.Errorf("failed to update status of service %s/%s: %v", namespace, name, err)	×
506	return err	×
507	}	×
508	}
509
510	return nil	×
511	}
512
513	func getVipIps(svc *v1.Service) []string {	×
514	var ips []string	×
515	if vip, ok := svc.Annotations[util.SwitchLBRuleVipsAnnotation]; ok {	×
516	ips = strings.Split(vip, ",")	×
517	} else {	×
518	ips = util.ServiceClusterIPs(*svc)	×
519	if svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] != "" {	×
520	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
521	ips = append(ips, ingress.IP)	×
522	}	×
523	}
524	}
525	return ips	×
526	}
527
528	func diffSvcPorts(oldPorts, newPorts []v1.ServicePort) (toDel []v1.ServicePort) {	×
529	for _, oldPort := range oldPorts {	×
530	found := false	×
531	for _, newPort := range newPorts {	×
532	if reflect.DeepEqual(oldPort, newPort) {	×
533	found = true	×
534	break	×
535	}
536	}
537	if !found {	×
538	toDel = append(toDel, oldPort)	×
539	}	×
540	}
541
542	return toDel	×
543	}
544
545	func (c Controller) checkServiceLBIPBelongToSubnet(svc v1.Service) error {	×
546	subnets, err := c.subnetsLister.List(labels.Everything())	×
547	if err != nil {	×
548	klog.Errorf("failed to list subnets: %v", err)	×
549	return err	×
550	}	×
551
552	isServiceExternalIPFromSubnet := false	×
553	for _, subnet := range subnets {	×
554	for _, ingress := range svc.Status.LoadBalancer.Ingress {	×
555	if util.CIDRContainIP(subnet.Spec.CIDRBlock, ingress.IP) {	×
556	svc.Annotations[util.ServiceExternalIPFromSubnetAnnotation] = subnet.Name	×
557	isServiceExternalIPFromSubnet = true	×
558	break	×
559	}
560	}
561	}
562
563	if !isServiceExternalIPFromSubnet {	×
564	delete(svc.Annotations, util.ServiceExternalIPFromSubnetAnnotation)	×
565	}	×
566	klog.Infof("Service %s/%s external IP belongs to subnet: %v", svc.Namespace, svc.Name, isServiceExternalIPFromSubnet)	×
567	if _, err = c.config.KubeClient.CoreV1().Services(svc.Namespace).Update(context.TODO(), svc, metav1.UpdateOptions{}); err != nil {	×
568	klog.Errorf("failed to update service %s/%s: %v", svc.Namespace, svc.Name, err)	×
569	return err	×
570	}	×
571
572	return nil	×
573	}

kubeovn / kube-ovn / 16863016720

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