kubeovn / kube-ovn / 17228354799

coverage: 21.341% (-0.2%) from 21.508%
17228354799

push

github

oilbeater 
handle delete final state unknown object in enqueue handler (#5649)

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

0 of 443 new or added lines in 27 files covered. (0.0%)

1 existing line in 1 file now uncovered.

10514 of 49267 relevant lines covered (21.34%)

0.25 hits per line

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

0.0
/pkg/controller/node.go
1 
package controller
2 









3



import (










4



        "context"










5



        "errors"










6



        "fmt"










7



        "reflect"










8



        "slices"










9



        "strconv"










10



        "strings"










11



        "time"










12












13



        goping "github.com/prometheus-community/pro-bing"










14



        "github.com/scylladb/go-set/strset"










15



        v1 "k8s.io/api/core/v1"










16



        k8serrors "k8s.io/apimachinery/pkg/api/errors"










17



        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"










18



        "k8s.io/apimachinery/pkg/labels"










19



        "k8s.io/apimachinery/pkg/types"










20



        "k8s.io/client-go/tools/cache"










21



        "k8s.io/klog/v2"










22












23



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










24



        "github.com/kubeovn/kube-ovn/pkg/ovs"










25



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










26



)










27












28



type ErrChassisNotFound struct {










29



        Chassis string










30



        Node    string










31



}










32












33



func (e *ErrChassisNotFound) Error() string {




×







34



        return fmt.Sprintf("chassis %s not found for node %s", e.Chassis, e.Node)




×







35



}




×







36












37



func (c *Controller) enqueueAddNode(obj any) {




×







38



        key := cache.MetaObjectToName(obj.(*v1.Node)).String()




×







39



        klog.V(3).Infof("enqueue add node %s", key)




×







40



        c.addNodeQueue.Add(key)




×







41



}




×







42












43



func nodeReady(node *v1.Node) bool {




×







44



        var ready, networkUnavailable bool




×







45



        for _, c := range node.Status.Conditions {




×







46



                switch c.Type {




×







47



                case v1.NodeReady:




×







48



                        ready = c.Status == v1.ConditionTrue




×







49



                case v1.NodeNetworkUnavailable:




×







50



                        networkUnavailable = c.Status == v1.ConditionTrue




×







51



                }










52



        }










53



        return ready && !networkUnavailable




×







54



}










55












56



func (c *Controller) enqueueUpdateNode(oldObj, newObj any) {




×







57



        oldNode := oldObj.(*v1.Node)




×







58



        newNode := newObj.(*v1.Node)




×







59






×







60



        if nodeReady(oldNode) != nodeReady(newNode) ||




×







61



                !reflect.DeepEqual(oldNode.Annotations, newNode.Annotations) {




×







62



                key := cache.MetaObjectToName(newNode).String()




×







63



                if len(newNode.Annotations) == 0 || newNode.Annotations[util.AllocatedAnnotation] != "true" {




×







64



                        klog.V(3).Infof("enqueue add node %s", key)




×







65



                        c.addNodeQueue.Add(key)




×







66



                } else {




×







67



                        klog.V(3).Infof("enqueue update node %s", key)




×







68



                        c.updateNodeQueue.Add(key)




×







69



                }




×







70



        }










71



}










72












73



func (c *Controller) enqueueDeleteNode(obj any) {




×








NEW


74



        var node *v1.Node




×








NEW


75



        switch t := obj.(type) {




×








NEW


76



        case *v1.Node:




×








NEW


77



                node = t




×








NEW


78



        case cache.DeletedFinalStateUnknown:




×








NEW


79



                n, ok := t.Obj.(*v1.Node)




×








NEW


80



                if !ok {




×








NEW


81



                        klog.Warningf("unexpected object type: %T", t.Obj)




×








NEW


82



                        return




×








NEW


83



                }




×








NEW


84



                node = n




×








NEW


85



        default:




×








NEW


86



                klog.Warningf("unexpected type: %T", obj)




×








NEW


87



                return




×







88



        }










89












90



        key := cache.MetaObjectToName(node).String()




×







91



        klog.V(3).Infof("enqueue delete node %s", key)




×







92



        c.deletingNodeObjMap.Store(key, node)




×







93



        c.deleteNodeQueue.Add(key)




×







94



}










95












96



func nodeUnderlayAddressSetName(node string, af int) string {




×







97



        return fmt.Sprintf("node_%s_underlay_v%d", strings.ReplaceAll(node, "-", "_"), af)




×







98



}




×







99












100



func (c *Controller) handleAddNode(key string) error {




×







101



        c.nodeKeyMutex.LockKey(key)




×







102



        defer func() { _ = c.nodeKeyMutex.UnlockKey(key) }()




×







103












104



        cachedNode, err := c.nodesLister.Get(key)




×







105



        if err != nil {




×







106



                if k8serrors.IsNotFound(err) {




×







107



                        return nil




×







108



                }




×







109



                klog.Errorf("failed to get node %s: %v", key, err)




×







110



                return err




×







111



        }










112



        node := cachedNode.DeepCopy()




×







113



        klog.Infof("handle add node %s", node.Name)




×







114






×







115



        subnets, err := c.subnetsLister.List(labels.Everything())




×







116



        if err != nil {




×







117



                klog.Errorf("failed to list subnets: %v", err)




×







118



                return err




×







119



        }




×







120












121



        nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)




×







122



        for _, subnet := range subnets {




×







123



                if subnet.Spec.Vpc != c.config.ClusterRouter {




×







124



                        continue




×







125



                }










126












127



                v4, v6 := util.SplitStringIP(subnet.Spec.CIDRBlock)




×







128



                if subnet.Spec.Vlan == "" && (util.CIDRContainIP(v4, nodeIPv4) || util.CIDRContainIP(v6, nodeIPv6)) {




×







129



                        msg := fmt.Sprintf("internal IP address of node %s is in CIDR of subnet %s, this may result in network issues", node.Name, subnet.Name)




×







130



                        klog.Warning(msg)




×







131



                        c.recorder.Eventf(&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: node.Name, UID: types.UID(node.Name)}}, v1.EventTypeWarning, "NodeAddressConflictWithSubnet", msg)




×







132



                        break




×







133



                }










134



        }










135












136



        if err = c.handleNodeAnnotationsForProviderNetworks(node); err != nil {




×







137



                klog.Errorf("failed to handle annotations of node %s for provider networks: %v", node.Name, err)




×







138



                return err




×







139



        }




×







140












141



        subnet, err := c.subnetsLister.Get(c.config.NodeSwitch)




×







142



        if err != nil {




×







143



                klog.Errorf("failed to get node subnet: %v", err)




×







144



                return err




×







145



        }




×







146












147



        var v4IP, v6IP, mac string




×







148






×







149



        portName := util.NodeLspName(key)




×







150



        if node.Annotations[util.AllocatedAnnotation] == "true" && node.Annotations[util.IPAddressAnnotation] != "" && node.Annotations[util.MacAddressAnnotation] != "" {




×







151



                macStr := node.Annotations[util.MacAddressAnnotation]




×







152



                v4IP, v6IP, mac, err = c.ipam.GetStaticAddress(portName, portName, node.Annotations[util.IPAddressAnnotation],




×







153



                        &macStr, node.Annotations[util.LogicalSwitchAnnotation], true)




×







154



                if err != nil {




×







155



                        klog.Errorf("failed to alloc static ip addrs for node %v: %v", node.Name, err)




×







156



                        return err




×







157



                }




×







158



        } else {




×







159



                v4IP, v6IP, mac, err = c.ipam.GetRandomAddress(portName, portName, nil, c.config.NodeSwitch, "", nil, true)




×







160



                if err != nil {




×







161



                        klog.Errorf("failed to alloc random ip addrs for node %v: %v", node.Name, err)




×







162



                        return err




×







163



                }




×







164












165



                // Clean up potentially existing logical switch ports to avoid leftover issues from previous failed configurations










166



                if err := c.OVNNbClient.DeleteLogicalSwitchPort(portName); err != nil {




×







167



                        klog.Errorf("failed to delete stale logical switch port %s: %v", portName, err)




×







168



                        return err




×







169



                }




×







170



                klog.Infof("deleted stale logical switch port %s", portName)




×







171



        }










172












173



        ipStr := util.GetStringIP(v4IP, v6IP)




×







174



        if err := c.OVNNbClient.CreateBareLogicalSwitchPort(c.config.NodeSwitch, portName, ipStr, mac); err != nil {




×







175



                klog.Errorf("failed to create logical switch port %s: %v", portName, err)




×







176



                return err




×







177



        }




×







178












179



        for ip := range strings.SplitSeq(ipStr, ",") {




×







180



                if ip == "" {




×







181



                        continue




×







182



                }










183












184



                nodeIP, af := nodeIPv4, 4




×







185



                protocol := util.CheckProtocol(ip)




×







186



                if protocol == kubeovnv1.ProtocolIPv6 {




×







187



                        nodeIP, af = nodeIPv6, 6




×







188



                }




×







189



                if nodeIP != "" {




×







190



                        var (




×







191



                                match       = fmt.Sprintf("ip%d.dst == %s", af, nodeIP)




×







192



                                action      = kubeovnv1.PolicyRouteActionReroute




×







193



                                externalIDs = map[string]string{




×







194



                                        "vendor":         util.CniTypeName,




×







195



                                        "node":           node.Name,




×







196



                                        "address-family": strconv.Itoa(af),




×







197



                                }




×







198



                        )




×







199



                        klog.Infof("add policy route for router: %s, match %s, action %s, nexthop %s, externalID %v", c.config.ClusterRouter, match, action, ip, externalIDs)




×







200



                        if err = c.addPolicyRouteToVpc(




×







201



                                c.config.ClusterRouter,




×







202



                                &kubeovnv1.PolicyRoute{




×







203



                                        Priority:  util.NodeRouterPolicyPriority,




×







204



                                        Match:     match,




×







205



                                        Action:    action,




×







206



                                        NextHopIP: ip,




×







207



                                },




×







208



                                externalIDs,




×







209



                        ); err != nil {




×







210



                                klog.Errorf("failed to add logical router policy for node %s: %v", node.Name, err)




×







211



                                return err




×







212



                        }




×







213












214



                        dnsIPs := make([]string, 0, len(c.config.NodeLocalDNSIPs))




×







215



                        for _, ip := range c.config.NodeLocalDNSIPs {




×







216



                                if util.CheckProtocol(ip) == protocol {




×







217



                                        dnsIPs = append(dnsIPs, ip)




×







218



                                }




×







219



                        }










220












221



                        if err = c.addPolicyRouteForLocalDNSCacheOnNode(dnsIPs, portName, ip, node.Name, af); err != nil {




×







222



                                klog.Errorf("failed to add policy route for node %s: %v", node.Name, err)




×







223



                                return err




×







224



                        }




×







225



                }










226



        }










227












228



        if err := c.addNodeGatewayStaticRoute(); err != nil {




×







229



                klog.Errorf("failed to add static route for node gw: %v", err)




×







230



                return err




×







231



        }




×







232












233



        patch := util.KVPatch{




×







234



                util.IPAddressAnnotation:     ipStr,




×







235



                util.MacAddressAnnotation:    mac,




×







236



                util.CidrAnnotation:          subnet.Spec.CIDRBlock,




×







237



                util.GatewayAnnotation:       subnet.Spec.Gateway,




×







238



                util.LogicalSwitchAnnotation: c.config.NodeSwitch,




×







239



                util.AllocatedAnnotation:     "true",




×







240



                util.PortNameAnnotation:      portName,




×







241



        }




×







242



        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Nodes(), node.Name, patch); err != nil {




×







243



                klog.Errorf("failed to update annotations of node %s: %v", node.Name, err)




×







244



                return err




×







245



        }




×







246












247



        if err := c.createOrUpdateIPCR("", "", ipStr, mac, c.config.NodeSwitch, "", node.Name, ""); err != nil {




×







248



                klog.Errorf("failed to create or update IPs %s: %v", portName, err)




×







249



                return err




×







250



        }




×







251












252



        for _, subnet := range subnets {




×







253



                if (subnet.Spec.Vlan != "" && !subnet.Spec.LogicalGateway) || subnet.Spec.Vpc != c.config.ClusterRouter || subnet.Name == c.config.NodeSwitch || subnet.Spec.GatewayType != kubeovnv1.GWDistributedType {




×







254



                        continue




×







255



                }










256



                if err = c.createPortGroupForDistributedSubnet(node, subnet); err != nil {




×







257



                        klog.Errorf("failed to create port group for node %s and subnet %s: %v", node.Name, subnet.Name, err)




×







258



                        return err




×







259



                }




×







260



                // policy route for overlay distributed subnet should be reconciled when node ip changed










261



                c.addOrUpdateSubnetQueue.Add(subnet.Name)




×







262



        }










263












264



        // ovn acl doesn't support address_set name with '-', so replace '-' by '.'










265



        pgName := strings.ReplaceAll(portName, "-", ".")




×







266



        if err = c.OVNNbClient.CreatePortGroup(pgName, map[string]string{"node": node.Name, networkPolicyKey: "node" + "/" + key}); err != nil {




×







267



                klog.Errorf("create port group %s for node %s: %v", pgName, key, err)




×







268



                return err




×







269



        }




×







270












271



        if err := c.addPolicyRouteForCentralizedSubnetOnNode(node.Name, ipStr); err != nil {




×







272



                klog.Errorf("failed to add policy route for node %s, %v", key, err)




×







273



                return err




×







274



        }




×







275












276



        if err := c.UpdateChassisTag(node); err != nil {




×







277



                klog.Errorf("failed to update chassis tag for node %s: %v", node.Name, err)




×







278



                return err




×







279



        }




×







280












281



        if err := c.retryDelDupChassis(util.ChassisRetryMaxTimes, util.ChassisControllerRetryInterval, c.cleanDuplicatedChassis, node); err != nil {




×







282



                klog.Errorf("failed to clean duplicated chassis for node %s: %v", node.Name, err)




×







283



                return err




×







284



        }




×







285



        return nil




×







286



}










287












288



func (c *Controller) handleNodeAnnotationsForProviderNetworks(node *v1.Node) error {




×







289



        providerNetworks, err := c.providerNetworksLister.List(labels.Everything())




×







290



        if err != nil && !k8serrors.IsNotFound(err) {




×







291



                klog.Errorf("failed to list provider networks: %v", err)




×







292



                return err




×







293



        }




×







294












295



        for _, pn := range providerNetworks {




×







296



                excludeAnno := fmt.Sprintf(util.ProviderNetworkExcludeTemplate, pn.Name)




×







297



                interfaceAnno := fmt.Sprintf(util.ProviderNetworkInterfaceTemplate, pn.Name)




×







298






×







299



                var newPn *kubeovnv1.ProviderNetwork




×







300



                excluded := slices.Contains(pn.Spec.ExcludeNodes, node.Name)




×







301



                if !excluded && len(node.Annotations) != 0 && node.Annotations[excludeAnno] == "true" {




×







302



                        newPn = pn.DeepCopy()




×







303



                        newPn.Spec.ExcludeNodes = append(newPn.Spec.ExcludeNodes, node.Name)




×







304



                        excluded = true




×







305



                }




×







306












307



                var customInterface string




×







308



                for _, v := range pn.Spec.CustomInterfaces {




×







309



                        if slices.Contains(v.Nodes, node.Name) {




×







310



                                customInterface = v.Interface




×







311



                                break




×







312



                        }










313



                }










314



                if customInterface == "" && len(node.Annotations) != 0 {




×







315



                        if customInterface = node.Annotations[interfaceAnno]; customInterface != "" {




×







316



                                if newPn == nil {




×







317



                                        newPn = pn.DeepCopy()




×







318



                                }




×







319



                                var index *int




×







320



                                for i := range newPn.Spec.CustomInterfaces {




×







321



                                        if newPn.Spec.CustomInterfaces[i].Interface == customInterface {




×







322



                                                index = &i




×







323



                                                break




×







324



                                        }










325



                                }










326



                                if index != nil {




×







327



                                        newPn.Spec.CustomInterfaces[*index].Nodes = append(newPn.Spec.CustomInterfaces[*index].Nodes, node.Name)




×







328



                                } else {




×







329



                                        ci := kubeovnv1.CustomInterface{Interface: customInterface, Nodes: []string{node.Name}}




×







330



                                        newPn.Spec.CustomInterfaces = append(newPn.Spec.CustomInterfaces, ci)




×







331



                                }




×







332



                        }










333



                }










334












335



                if newPn != nil {




×







336



                        if newPn, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().Update(context.Background(), newPn, metav1.UpdateOptions{}); err != nil {




×







337



                                klog.Errorf("failed to update provider network %s: %v", pn.Name, err)




×







338



                                return err




×







339



                        }




×







340



                }










341












342



                if len(node.Annotations) != 0 {




×







343



                        patch := util.KVPatch{excludeAnno: nil, interfaceAnno: nil}




×







344



                        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Nodes(), node.Name, patch); err != nil {




×







345



                                klog.Errorf("failed to patch node %s: %v", node.Name, err)




×







346



                                return err




×







347



                        }




×







348



                }










349












350



                if excluded {




×







351



                        if newPn == nil {




×







352



                                newPn = pn.DeepCopy()




×







353



                        } else {




×







354



                                newPn = newPn.DeepCopy()




×







355



                        }




×







356












357



                        if newPn.Status.EnsureNodeStandardConditions(node.Name) {




×







358



                                _, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().UpdateStatus(context.Background(), newPn, metav1.UpdateOptions{})




×







359



                                if err != nil {




×







360



                                        klog.Errorf("failed to update status of provider network %s: %v", pn.Name, err)




×







361



                                        return err




×







362



                                }




×







363



                        }










364



                }










365



        }










366












367



        return nil




×







368



}










369












370



func (c *Controller) handleDeleteNode(key string) (err error) {




×







371



        c.nodeKeyMutex.LockKey(key)




×







372



        defer func() {




×







373



                _ = c.nodeKeyMutex.UnlockKey(key)




×







374



                if err == nil {




×







375



                        c.deletingNodeObjMap.Delete(key)




×







376



                }




×







377



        }()










378



        klog.Infof("handle delete node %s", key)




×







379






×







380



        node, ok := c.deletingNodeObjMap.Load(key)




×







381



        if !ok {




×







382



                return nil




×







383



        }




×







384



        n, _ := c.nodesLister.Get(key)




×







385



        if n != nil && n.UID != node.UID {




×







386



                klog.Warningf("Node %s is adding, skip the node delete handler, but it may leave some gc resources behind", key)




×







387



                return nil




×







388



        }




×







389



        return c.deleteNode(key)




×







390



}










391












392



func (c *Controller) deleteNode(key string) error {




×







393



        portName := util.NodeLspName(key)




×







394



        klog.Infof("delete logical switch port %s", portName)




×







395



        if err := c.OVNNbClient.DeleteLogicalSwitchPort(portName); err != nil {




×







396



                klog.Errorf("failed to delete node switch port %s: %v", portName, err)




×







397



                return err




×







398



        }




×







399



        if err := c.OVNSbClient.DeleteChassisByHost(key); err != nil {




×







400



                klog.Errorf("failed to delete chassis for node %s: %v", key, err)




×







401



                return err




×







402



        }




×







403












404



        for _, af := range [...]int{4, 6} {




×







405



                if err := c.deletePolicyRouteForLocalDNSCacheOnNode(key, af); err != nil {




×







406



                        klog.Error(err)




×







407



                        return err




×







408



                }




×







409



        }










410












411



        // ovn acl doesn't support address_set name with '-', so replace '-' by '.'










412



        pgName := strings.ReplaceAll(portName, "-", ".")




×







413



        if err := c.OVNNbClient.DeletePortGroup(pgName); err != nil {




×







414



                klog.Errorf("delete port group %s for node: %v", portName, err)




×







415



                return err




×







416



        }




×







417












418



        if err := c.deletePolicyRouteForNode(key, portName); err != nil {




×







419



                klog.Errorf("failed to delete policy route for node %s: %v", key, err)




×







420



                return err




×







421



        }




×







422












423



        if err := c.OVNNbClient.DeleteAddressSet(nodeUnderlayAddressSetName(key, 4)); err != nil {




×







424



                klog.Errorf("failed to delete address set for node %s: %v", key, err)




×







425



                return err




×







426



        }




×







427



        if err := c.OVNNbClient.DeleteAddressSet(nodeUnderlayAddressSetName(key, 6)); err != nil {




×







428



                klog.Errorf("failed to delete address set for node %s: %v", key, err)




×







429



                return err




×







430



        }




×







431












432



        klog.Infof("release node port %s", portName)




×







433



        c.ipam.ReleaseAddressByPod(portName, c.config.NodeSwitch)




×







434






×







435



        providerNetworks, err := c.providerNetworksLister.List(labels.Everything())




×







436



        if err != nil && !k8serrors.IsNotFound(err) {




×







437



                klog.Errorf("failed to list provider networks: %v", err)




×







438



                return err




×







439



        }




×







440












441



        for _, pn := range providerNetworks {




×







442



                if err = c.updateProviderNetworkForNodeDeletion(pn, key); err != nil {




×







443



                        klog.Error(err)




×







444



                        return err




×







445



                }




×







446



        }










447



        klog.Infof("delete node ip %s", portName)




×







448



        if err = c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), portName, metav1.DeleteOptions{}); err != nil && !k8serrors.IsNotFound(err) {




×







449



                return err




×







450



        }




×







451












452



        return nil




×







453



}










454












455



func (c *Controller) updateProviderNetworkForNodeDeletion(pn *kubeovnv1.ProviderNetwork, node string) error {




×







456



        // update provider network status




×







457



        var needUpdate bool




×







458



        newPn := pn.DeepCopy()




×







459



        if slices.Contains(newPn.Status.ReadyNodes, node) {




×







460



                newPn.Status.ReadyNodes = util.RemoveString(newPn.Status.ReadyNodes, node)




×







461



                needUpdate = true




×







462



        }




×







463



        if newPn.Status.RemoveNodeConditions(node) {




×







464



                needUpdate = true




×







465



        }




×







466



        if needUpdate {




×







467



                var err error




×







468



                newPn, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().UpdateStatus(context.Background(), newPn, metav1.UpdateOptions{})




×







469



                if err != nil {




×







470



                        klog.Errorf("failed to update status of provider network %s: %v", pn.Name, err)




×







471



                        return err




×







472



                }




×







473



        }










474












475



        // update provider network spec










476



        pn, newPn = newPn, nil




×







477



        if excludeNodes := util.RemoveString(pn.Spec.ExcludeNodes, node); len(excludeNodes) != len(pn.Spec.ExcludeNodes) {




×







478



                newPn = pn.DeepCopy()




×







479



                newPn.Spec.ExcludeNodes = excludeNodes




×







480



        }




×







481












482



        var changed bool




×







483



        customInterfaces := make([]kubeovnv1.CustomInterface, 0, len(pn.Spec.CustomInterfaces))




×







484



        for _, ci := range pn.Spec.CustomInterfaces {




×







485



                nodes := util.RemoveString(ci.Nodes, node)




×







486



                if !changed {




×







487



                        changed = len(nodes) == 0 || len(nodes) != len(ci.Nodes)




×







488



                }




×







489



                if len(nodes) != 0 {




×







490



                        customInterfaces = append(customInterfaces, kubeovnv1.CustomInterface{Interface: ci.Interface, Nodes: nodes})




×







491



                }




×







492



        }










493



        if changed {




×







494



                newPn = pn.DeepCopy()




×







495



                newPn.Spec.CustomInterfaces = customInterfaces




×







496



        }




×







497



        if newPn != nil {




×







498



                if _, err := c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().Update(context.Background(), newPn, metav1.UpdateOptions{}); err != nil {




×







499



                        klog.Errorf("failed to update provider network %s: %v", pn.Name, err)




×







500



                        return err




×







501



                }




×







502



        }










503












504



        return nil




×







505



}










506












507



func (c *Controller) handleUpdateNode(key string) error {




×







508



        c.nodeKeyMutex.LockKey(key)




×







509



        defer func() { _ = c.nodeKeyMutex.UnlockKey(key) }()




×







510



        klog.Infof("handle update node %s", key)




×







511






×







512



        node, err := c.nodesLister.Get(key)




×







513



        if err != nil {




×







514



                if k8serrors.IsNotFound(err) {




×







515



                        return nil




×







516



                }




×







517



                klog.Errorf("failed to get node %s: %v", key, err)




×







518



                return err




×







519



        }










520












521



        if err = c.handleNodeAnnotationsForProviderNetworks(node); err != nil {




×







522



                klog.Errorf("failed to handle annotations of node %s for provider networks: %v", node.Name, err)




×







523



                return err




×







524



        }




×







525












526



        subnets, err := c.subnetsLister.List(labels.Everything())




×







527



        if err != nil {




×







528



                klog.Errorf("failed to get subnets %v", err)




×







529



                return err




×







530



        }




×







531












532



        if err := c.UpdateChassisTag(node); err != nil {




×







533



                klog.Errorf("failed to update chassis tag for node %s: %v", node.Name, err)




×







534



                return err




×







535



        }




×







536



        if err := c.retryDelDupChassis(util.ChassisRetryMaxTimes, util.ChassisControllerRetryInterval, c.cleanDuplicatedChassis, node); err != nil {




×







537



                klog.Errorf("failed to clean duplicated chassis for node %s: %v", node.Name, err)




×







538



                return err




×







539



        }




×







540












541



        for _, cachedSubnet := range subnets {




×







542



                if cachedSubnet.Spec.GatewayType == kubeovnv1.GWDistributedType {




×







543



                        // we need to reconcile ovn route for subnets with distributed gateway mode,




×







544



                        // since the informer node cache may not be synced before the subnet reconcile triggered by node addition




×







545



                        c.addOrUpdateSubnetQueue.Add(cachedSubnet.Name)




×







546



                        continue




×







547



                }










548



                subnet := cachedSubnet.DeepCopy()




×







549



                if util.GatewayContains(subnet.Spec.GatewayNode, node.Name) {




×







550



                        if err := c.reconcileOvnDefaultVpcRoute(subnet); err != nil {




×







551



                                klog.Error(err)




×







552



                                return err




×







553



                        }




×







554



                }










555



        }










556












557



        return nil




×







558



}










559












560



func (c *Controller) checkSubnetGateway() {




×







561



        if err := c.checkSubnetGatewayNode(); err != nil {




×







562



                klog.Errorf("failed to check subnet gateway node: %v", err)




×







563



        }




×







564



}










565












566



func (c *Controller) checkSubnetGatewayNode() error {




×







567



        klog.V(3).Infoln("start to check subnet gateway node")




×







568



        subnetList, err := c.subnetsLister.List(labels.Everything())




×







569



        if err != nil {




×







570



                klog.Errorf("failed to list subnets: %v", err)




×







571



                return err




×







572



        }




×







573



        nodes, err := c.nodesLister.List(labels.Everything())




×







574



        if err != nil {




×







575



                klog.Errorf("failed to list nodes: %v", err)




×







576



                return err




×







577



        }




×







578












579



        for _, subnet := range subnetList {




×







580



                if (subnet.Spec.Vlan != "" && (subnet.Spec.U2OInterconnection || !subnet.Spec.LogicalGateway)) ||




×







581



                        subnet.Spec.Vpc != c.config.ClusterRouter ||




×







582



                        subnet.Name == c.config.NodeSwitch ||




×







583



                        subnet.Spec.GatewayNode == "" ||




×







584



                        subnet.Spec.GatewayType != kubeovnv1.GWCentralizedType ||




×







585



                        !subnet.Spec.EnableEcmp {




×







586



                        continue




×







587



                }










588



                gwNodes := strings.Split(subnet.Spec.GatewayNode, ",")




×







589



                if len(gwNodes) < 2 {




×







590



                        continue




×







591



                }










592












593



                for cidrBlock := range strings.SplitSeq(subnet.Spec.CIDRBlock, ",") {




×







594



                        nextHops, nameIPMap, err := c.getPolicyRouteParas(cidrBlock, util.GatewayRouterPolicyPriority)




×







595



                        if err != nil {




×







596



                                klog.Errorf("failed to get ecmp policy route paras for subnet %s: %v", subnet.Name, err)




×







597



                                continue




×







598



                        }










599



                        for _, node := range nodes {




×







600



                                ipStr := node.Annotations[util.IPAddressAnnotation]




×







601



                                for ip := range strings.SplitSeq(ipStr, ",") {




×







602



                                        if util.CheckProtocol(cidrBlock) != util.CheckProtocol(ip) {




×







603



                                                continue




×







604



                                        }










605












606



                                        exist := nameIPMap[node.Name] == ip




×







607



                                        if util.GatewayContains(subnet.Spec.GatewayNode, node.Name) {




×







608



                                                pinger, err := goping.NewPinger(ip)




×







609



                                                if err != nil {




×







610



                                                        return fmt.Errorf("failed to init pinger, %w", err)




×







611



                                                }




×







612



                                                pinger.SetPrivileged(true)




×







613






×







614



                                                count := 5




×







615



                                                pinger.Count = count




×







616



                                                pinger.Timeout = time.Duration(count) * time.Second




×







617



                                                pinger.Interval = 1 * time.Second




×







618






×







619



                                                var pingSucceeded bool




×







620



                                                pinger.OnRecv = func(_ *goping.Packet) {




×







621



                                                        pingSucceeded = true




×







622



                                                        pinger.Stop()




×







623



                                                }




×







624



                                                if err = pinger.Run(); err != nil {




×







625



                                                        klog.Errorf("failed to run pinger for destination %s: %v", ip, err)




×







626



                                                        return err




×







627



                                                }




×







628












629



                                                nodeIsReady := nodeReady(node)




×







630



                                                if !pingSucceeded || !nodeIsReady {




×







631



                                                        if exist {




×







632



                                                                if !pingSucceeded {




×







633



                                                                        klog.Warningf("failed to ping ovn0 ip %s on node %s", ip, node.Name)




×







634



                                                                }




×







635



                                                                if !nodeIsReady {




×







636



                                                                        klog.Warningf("node %s is not ready", node.Name)




×







637



                                                                }




×







638



                                                                klog.Warningf("delete ecmp policy route for node %s ip %s", node.Name, ip)




×







639



                                                                nextHops.Remove(ip)




×







640



                                                                delete(nameIPMap, node.Name)




×







641



                                                                klog.Infof("update policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







642



                                                                if err = c.updatePolicyRouteForCentralizedSubnet(subnet.Name, cidrBlock, nextHops.List(), nameIPMap); err != nil {




×







643



                                                                        klog.Errorf("failed to delete ecmp policy route for subnet %s on node %s, %v", subnet.Name, node.Name, err)




×







644



                                                                        return err




×







645



                                                                }




×







646



                                                        }










647



                                                } else {




×







648



                                                        klog.V(3).Infof("succeeded to ping ovn0 ip %s on node %s", ip, node.Name)




×







649



                                                        if !exist {




×







650



                                                                nextHops.Add(ip)




×







651



                                                                if nameIPMap == nil {




×







652



                                                                        nameIPMap = make(map[string]string, 1)




×







653



                                                                }




×







654



                                                                nameIPMap[node.Name] = ip




×







655



                                                                klog.Infof("update policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







656



                                                                if err = c.updatePolicyRouteForCentralizedSubnet(subnet.Name, cidrBlock, nextHops.List(), nameIPMap); err != nil {




×







657



                                                                        klog.Errorf("failed to add ecmp policy route for subnet %s on node %s, %v", subnet.Name, node.Name, err)




×







658



                                                                        return err




×







659



                                                                }




×







660



                                                        }










661



                                                }










662



                                        } else if exist {




×







663



                                                klog.Infof("subnet %s gateway nodes does not contain node %s, delete policy route for node ip %s", subnet.Name, node.Name, ip)




×







664



                                                nextHops.Remove(ip)




×







665



                                                delete(nameIPMap, node.Name)




×







666



                                                klog.Infof("update policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







667



                                                if err = c.updatePolicyRouteForCentralizedSubnet(subnet.Name, cidrBlock, nextHops.List(), nameIPMap); err != nil {




×







668



                                                        klog.Errorf("failed to delete ecmp policy route for subnet %s on node %s, %v", subnet.Name, node.Name, err)




×







669



                                                        return err




×







670



                                                }




×







671



                                        }










672



                                }










673



                        }










674



                }










675



        }










676



        return nil




×







677



}










678












679



func (c *Controller) cleanDuplicatedChassis(node *v1.Node) error {




×







680



        // if multi chassis has the same node name, delete all of them




×







681



        var err error




×







682



        if _, err := c.OVNSbClient.GetChassisByHost(node.Name); err == nil {




×







683



                return nil




×







684



        }




×







685



        klog.Errorf("failed to get chassis for node %s: %v", node.Name, err)




×







686



        if errors.Is(err, ovs.ErrOneNodeMultiChassis) {




×







687



                klog.Warningf("node %s has multiple chassis", node.Name)




×







688



                if err := c.OVNSbClient.DeleteChassisByHost(node.Name); err != nil {




×







689



                        klog.Errorf("failed to delete chassis for node %s: %v", node.Name, err)




×







690



                        return err




×







691



                }




×







692



        }










693



        return err




×







694



}










695












696



func (c *Controller) retryDelDupChassis(attempts, sleep int, f func(node *v1.Node) error, node *v1.Node) (err error) {




×







697



        i := 0




×







698



        for ; ; i++ {




×







699



                err = f(node)




×







700



                if err == nil {




×







701



                        return




×







702



                }




×







703



                klog.Errorf("failed to delete duplicated chassis for node %s: %v", node.Name, err)




×







704



                if i >= (attempts - 1) {




×







705



                        break




×







706



                }










707



                time.Sleep(time.Duration(sleep) * time.Second)




×







708



        }










709



        if i >= (attempts - 1) {




×







710



                errMsg := errors.New("exhausting all attempts")




×







711



                klog.Error(errMsg)




×







712



                return errMsg




×







713



        }




×







714



        klog.V(3).Infof("finish check chassis")




×







715



        return nil




×







716



}










717












718



func (c *Controller) fetchPodsOnNode(nodeName string, pods []*v1.Pod) ([]string, error) {




×







719



        ports := make([]string, 0, len(pods))




×







720



        for _, pod := range pods {




×







721



                if !isPodAlive(pod) || pod.Spec.HostNetwork || pod.Spec.NodeName != nodeName {




×







722



                        continue




×







723



                }










724












725



                if pod.Annotations[util.LogicalRouterAnnotation] != c.config.ClusterRouter {




×







726



                        subnetName := pod.Annotations[util.LogicalSwitchAnnotation]




×







727



                        if subnetName == "" {




×







728



                                klog.V(4).Infof("Pod %s/%s is not on cluster router and has no logical switch annotation, skipping for VLAN check.", pod.Namespace, pod.Name)




×







729



                                continue




×







730



                        }










731












732



                        subnet, err := c.subnetsLister.Get(subnetName)




×







733



                        if err != nil {




×







734



                                klog.Errorf("failed to get subnet %s: %v", subnetName, err)




×







735



                                return nil, err




×







736



                        }




×







737












738



                        if subnet.Spec.Vlan == "" {




×







739



                                continue




×







740



                        }










741



                }










742












743



                podName := c.getNameByPod(pod)




×







744






×







745



                podNets, err := c.getPodKubeovnNets(pod)




×







746



                if err != nil {




×







747



                        klog.Errorf("failed to get pod nets %v", err)




×







748



                        return nil, err




×







749



                }




×







750












751



                for _, podNet := range podNets {




×







752



                        if !isOvnSubnet(podNet.Subnet) {




×







753



                                continue




×







754



                        }










755












756



                        if pod.Annotations != nil && pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "true" {




×







757



                                ports = append(ports, ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName))




×







758



                        }




×







759



                }










760



        }










761



        return ports, nil




×







762



}










763












764



func (c *Controller) CheckNodePortGroup() {




×







765



        if err := c.checkAndUpdateNodePortGroup(); err != nil {




×







766



                klog.Errorf("check node port group status: %v", err)




×







767



        }




×







768



}










769












770



func (c *Controller) checkAndUpdateNodePortGroup() error {




×







771



        klog.V(3).Infoln("start to check node port-group status")




×







772



        var networkPolicyExists bool




×







773



        if c.config.EnableNP {




×







774



                np, _ := c.npsLister.List(labels.Everything())




×







775



                networkPolicyExists = len(np) != 0




×







776



        }




×







777












778



        nodes, err := c.nodesLister.List(labels.Everything())




×







779



        if err != nil {




×







780



                klog.Errorf("list nodes: %v", err)




×







781



                return err




×







782



        }




×







783












784



        pods, err := c.podsLister.List(labels.Everything())




×







785



        if err != nil {




×







786



                klog.Errorf("list pods, %v", err)




×







787



                return err




×







788



        }




×







789












790



        for _, node := range nodes {




×







791



                // The port-group should already created when add node




×







792



                pgName := strings.ReplaceAll(node.Annotations[util.PortNameAnnotation], "-", ".")




×







793






×







794



                // use join IP only when no internal IP exists




×







795



                nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)




×







796



                joinIP := node.Annotations[util.IPAddressAnnotation]




×







797



                joinIPv4, joinIPv6 := util.SplitStringIP(joinIP)




×







798



                if nodeIPv4 == "" {




×







799



                        nodeIPv4 = joinIPv4




×







800



                }




×







801



                if nodeIPv6 == "" {




×







802



                        nodeIPv6 = joinIPv6




×







803



                }




×







804



                nodeIP := strings.Trim(fmt.Sprintf("%s,%s", nodeIPv4, nodeIPv6), ",")




×







805






×







806



                nodePorts, err := c.fetchPodsOnNode(node.Name, pods)




×







807



                if err != nil {




×







808



                        klog.Errorf("fetch pods for node %v: %v", node.Name, err)




×







809



                        return err




×







810



                }




×







811












812



                if err = c.OVNNbClient.PortGroupSetPorts(pgName, nodePorts); err != nil {




×







813



                        klog.Errorf("failed to set ports of port group %s: %v", pgName, err)




×







814



                        return err




×







815



                }




×







816












817



                if networkPolicyExists {




×







818



                        if err := c.OVNNbClient.CreateNodeACL(pgName, nodeIP, joinIP); err != nil {




×







819



                                klog.Errorf("create node acl for node pg %s: %v", pgName, err)




×







820



                        }




×







821



                } else {




×







822



                        // clear all acl




×







823



                        if err = c.OVNNbClient.DeleteAcls(pgName, portGroupKey, "", nil); err != nil {




×







824



                                klog.Errorf("delete node acl for node pg %s: %v", pgName, err)




×







825



                        }




×







826



                }










827



        }










828












829



        return nil




×







830



}










831












832



func (c *Controller) UpdateChassisTag(node *v1.Node) error {




×







833



        annoChassisName := node.Annotations[util.ChassisAnnotation]




×







834



        if annoChassisName == "" {




×







835



                // kube-ovn-cni not ready to set chassis




×







836



                return nil




×







837



        }




×







838



        chassis, err := c.OVNSbClient.GetChassis(annoChassisName, true)




×







839



        if err != nil {




×







840



                klog.Errorf("failed to get chassis %s for node %s: %v", annoChassisName, node.Name, err)




×







841



                return err




×







842



        }




×







843



        if chassis == nil {




×







844



                klog.Infof("chassis %q not registered for node %s, do chassis gc once", annoChassisName, node.Name)




×







845



                // chassis name conflict, do GC




×







846



                if err = c.gcChassis(); err != nil {




×







847



                        klog.Errorf("failed to gc chassis: %v", err)




×







848



                        return err




×







849



                }




×







850



                err = &ErrChassisNotFound{Chassis: annoChassisName, Node: node.Name}




×







851



                klog.Error(err)




×







852



                return err




×







853



        }










854












855



        if chassis.ExternalIDs == nil || chassis.ExternalIDs["vendor"] != util.CniTypeName {




×







856



                klog.Infof("init tag %s for node %s chassis %s", util.CniTypeName, node.Name, chassis.Name)




×







857



                if err = c.OVNSbClient.UpdateChassisTag(chassis.Name, node.Name); err != nil {




×







858



                        err := fmt.Errorf("failed to init chassis tag, %w", err)




×







859



                        klog.Error(err)




×







860



                        return err




×







861



                }




×







862



        }










863



        return nil




×







864



}










865












866



func (c *Controller) addNodeGatewayStaticRoute() error {




×







867



        // If user not manage static route for default vpc, just add route about ovn-default to join




×







868



        if vpc, err := c.vpcsLister.Get(c.config.ClusterRouter); err != nil || vpc.Spec.StaticRoutes != nil {




×







869



                existRoute, err := c.OVNNbClient.ListLogicalRouterStaticRoutes(c.config.ClusterRouter, nil, nil, "", nil)




×







870



                if err != nil {




×







871



                        klog.Errorf("failed to get vpc %s static route list, %v", c.config.ClusterRouter, err)




×







872



                }




×







873



                if len(existRoute) != 0 {




×







874



                        klog.Infof("skip add static route for node gw")




×







875



                        return nil




×







876



                }




×







877



        }










878



        dstCidr := "0.0.0.0/0,::/0"




×







879



        for cidrBlock := range strings.SplitSeq(dstCidr, ",") {




×







880



                for nextHop := range strings.SplitSeq(c.config.NodeSwitchGateway, ",") {




×







881



                        if util.CheckProtocol(cidrBlock) != util.CheckProtocol(nextHop) {




×







882



                                continue




×







883



                        }










884












885



                        if err := c.addStaticRouteToVpc(




×







886



                                c.config.ClusterRouter,




×







887



                                &kubeovnv1.StaticRoute{




×







888



                                        Policy:     kubeovnv1.PolicyDst,




×







889



                                        CIDR:       cidrBlock,




×







890



                                        NextHopIP:  nextHop,




×







891



                                        RouteTable: util.MainRouteTable,




×







892



                                },




×







893



                        ); err != nil {




×







894



                                klog.Errorf("failed to add static route for node gw: %v", err)




×







895



                                return err




×







896



                        }




×







897



                }










898



        }










899



        return nil




×







900



}










901












902



func (c *Controller) getPolicyRouteParas(cidr string, priority int) (*strset.Set, map[string]string, error) {




×







903



        ipSuffix := "ip4"




×







904



        if util.CheckProtocol(cidr) == kubeovnv1.ProtocolIPv6 {




×







905



                ipSuffix = "ip6"




×







906



        }




×







907



        match := fmt.Sprintf("%s.src == %s", ipSuffix, cidr)




×







908



        policyList, err := c.OVNNbClient.GetLogicalRouterPolicy(c.config.ClusterRouter, priority, match, true)




×







909



        if err != nil {




×







910



                klog.Errorf("failed to get logical router policy: %v", err)




×







911



                return nil, nil, err




×







912



        }




×







913



        if len(policyList) == 0 {




×







914



                return strset.New(), map[string]string{}, nil




×







915



        }




×







916



        return strset.New(policyList[0].Nexthops...), policyList[0].ExternalIDs, nil




×







917



}










918












919



func (c *Controller) deletePolicyRouteForNode(nodeName, portName string) error {




×







920



        subnets, err := c.subnetsLister.List(labels.Everything())




×







921



        if err != nil {




×







922



                klog.Errorf("get subnets: %v", err)




×







923



                return err




×







924



        }




×







925












926



        addresses := c.ipam.GetPodAddress(portName)




×







927



        for _, addr := range addresses {




×







928



                if addr.IP == "" {




×







929



                        continue




×







930



                }










931



                klog.Infof("deleting logical router policy with nexthop %q from %s for node %s", addr.IP, c.config.ClusterRouter, nodeName)




×







932



                if err = c.OVNNbClient.DeleteLogicalRouterPolicyByNexthop(c.config.ClusterRouter, util.NodeRouterPolicyPriority, addr.IP); err != nil {




×







933



                        klog.Errorf("failed to delete logical router policy with nexthop %q from %s for node %s: %v", addr.IP, c.config.ClusterRouter, nodeName, err)




×







934



                        return err




×







935



                }




×







936



        }










937












938



        for _, subnet := range subnets {




×







939



                if (subnet.Spec.Vlan != "" && !subnet.Spec.LogicalGateway) || subnet.Spec.Vpc != c.config.ClusterRouter || subnet.Name == c.config.NodeSwitch {




×







940



                        continue




×







941



                }










942












943



                if subnet.Spec.GatewayType == kubeovnv1.GWDistributedType {




×







944



                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, nodeName)




×







945



                        if err = c.OVNNbClient.DeletePortGroup(pgName); err != nil {




×







946



                                klog.Errorf("delete port group for subnet %s and node %s: %v", subnet.Name, nodeName, err)




×







947



                                return err




×







948



                        }




×







949












950



                        klog.Infof("delete policy route for distributed subnet %s, node %s", subnet.Name, nodeName)




×







951



                        if err = c.deletePolicyRouteForDistributedSubnet(subnet, nodeName); err != nil {




×







952



                                klog.Errorf("delete policy route for subnet %s and node %s: %v", subnet.Name, nodeName, err)




×







953



                                return err




×







954



                        }




×







955



                }










956












957



                if subnet.Spec.GatewayType == kubeovnv1.GWCentralizedType {




×







958



                        if subnet.Spec.EnableEcmp {




×







959



                                for cidrBlock := range strings.SplitSeq(subnet.Spec.CIDRBlock, ",") {




×







960



                                        nextHops, nameIPMap, err := c.getPolicyRouteParas(cidrBlock, util.GatewayRouterPolicyPriority)




×







961



                                        if err != nil {




×







962



                                                klog.Errorf("get ecmp policy route paras for subnet %v, error %v", subnet.Name, err)




×







963



                                                continue




×







964



                                        }










965












966



                                        exist := false




×







967



                                        if _, ok := nameIPMap[nodeName]; ok {




×







968



                                                exist = true




×







969



                                        }




×







970












971



                                        if exist {




×







972



                                                nextHops.Remove(nameIPMap[nodeName])




×







973



                                                delete(nameIPMap, nodeName)




×







974






×







975



                                                if nextHops.Size() == 0 {




×







976



                                                        klog.Infof("delete policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







977



                                                        if err := c.deletePolicyRouteForCentralizedSubnet(subnet); err != nil {




×







978



                                                                klog.Errorf("failed to delete policy route for centralized subnet %s, %v", subnet.Name, err)




×







979



                                                                return err




×







980



                                                        }




×







981



                                                } else {




×







982



                                                        klog.Infof("update policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







983



                                                        if err = c.updatePolicyRouteForCentralizedSubnet(subnet.Name, cidrBlock, nextHops.List(), nameIPMap); err != nil {




×







984



                                                                klog.Errorf("failed to update policy route for subnet %s on node %s, %v", subnet.Name, nodeName, err)




×







985



                                                                return err




×







986



                                                        }




×







987



                                                }










988



                                        }










989



                                }










990



                        } else {




×







991



                                klog.Infof("reconcile policy route for centralized subnet %s", subnet.Name)




×







992



                                if err := c.reconcileDefaultCentralizedSubnetRouteInDefaultVpc(subnet); err != nil {




×







993



                                        klog.Errorf("failed to delete policy route for centralized subnet %s, %v", subnet.Name, err)




×







994



                                        return err




×







995



                                }




×







996



                        }










997



                }










998



        }










999



        return nil




×







1000



}










1001












1002



func (c *Controller) addPolicyRouteForCentralizedSubnetOnNode(nodeName, nodeIP string) error {




×







1003



        subnets, err := c.subnetsLister.List(labels.Everything())




×







1004



        if err != nil {




×







1005



                klog.Errorf("failed to get subnets %v", err)




×







1006



                return err




×







1007



        }




×







1008












1009



        for _, subnet := range subnets {




×







1010



                if (subnet.Spec.Vlan != "" && !subnet.Spec.LogicalGateway) || subnet.Spec.Vpc != c.config.ClusterRouter || subnet.Name == c.config.NodeSwitch || subnet.Spec.GatewayType != kubeovnv1.GWCentralizedType {




×







1011



                        continue




×







1012



                }










1013












1014



                if subnet.Spec.EnableEcmp {




×







1015



                        if !util.GatewayContains(subnet.Spec.GatewayNode, nodeName) {




×







1016



                                continue




×







1017



                        }










1018












1019



                        for nextHop := range strings.SplitSeq(nodeIP, ",") {




×







1020



                                for cidrBlock := range strings.SplitSeq(subnet.Spec.CIDRBlock, ",") {




×







1021



                                        if util.CheckProtocol(cidrBlock) != util.CheckProtocol(nextHop) {




×







1022



                                                continue




×







1023



                                        }










1024












1025



                                        nextHops, nameIPMap, err := c.getPolicyRouteParas(cidrBlock, util.GatewayRouterPolicyPriority)




×







1026



                                        if err != nil {




×







1027



                                                klog.Errorf("get ecmp policy route paras for subnet %v, error %v", subnet.Name, err)




×







1028



                                                continue




×







1029



                                        }










1030



                                        if nameIPMap[nodeName] == nextHop {




×







1031



                                                continue




×







1032



                                        }










1033












1034



                                        nextHops.Add(nextHop)




×







1035



                                        if nameIPMap == nil {




×







1036



                                                nameIPMap = make(map[string]string, 1)




×







1037



                                        }




×







1038



                                        nameIPMap[nodeName] = nextHop




×







1039



                                        klog.Infof("update policy route for centralized subnet %s, nextHops %s", subnet.Name, nextHops)




×







1040



                                        if err = c.updatePolicyRouteForCentralizedSubnet(subnet.Name, cidrBlock, nextHops.List(), nameIPMap); err != nil {




×







1041



                                                klog.Errorf("failed to update policy route for subnet %s on node %s, %v", subnet.Name, nodeName, err)




×







1042



                                                return err




×







1043



                                        }




×







1044



                                }










1045



                        }










1046



                } else {




×







1047



                        if subnet.Status.ActivateGateway != nodeName {




×







1048



                                continue




×







1049



                        }










1050



                        klog.Infof("add policy route for centralized subnet %s, on node %s, ip %s", subnet.Name, nodeName, nodeIP)




×







1051



                        if err = c.addPolicyRouteForCentralizedSubnet(subnet, nodeName, nil, strings.Split(nodeIP, ",")); err != nil {




×







1052



                                klog.Errorf("failed to add active-backup policy route for centralized subnet %s: %v", subnet.Name, err)




×







1053



                                return err




×







1054



                        }




×







1055



                }










1056



        }










1057



        return nil




×







1058



}










1059












1060



func (c *Controller) addPolicyRouteForLocalDNSCacheOnNode(dnsIPs []string, nodePortName, nodeIP, nodeName string, af int) error {




×







1061



        if len(dnsIPs) == 0 {




×







1062



                return c.deletePolicyRouteForLocalDNSCacheOnNode(nodeName, af)




×







1063



        }




×







1064












1065



        var (




×







1066



                externalIDs = map[string]string{




×







1067



                        "vendor":          util.CniTypeName,




×







1068



                        "node":            nodeName,




×







1069



                        "address-family":  strconv.Itoa(af),




×







1070



                        "isLocalDnsCache": "true",




×







1071



                }




×







1072



                pgAs     = strings.ReplaceAll(fmt.Sprintf("%s_ip%d", nodePortName, af), "-", ".")




×







1073



                action   = kubeovnv1.PolicyRouteActionReroute




×







1074



                nextHops = []string{nodeIP}




×







1075



        )




×







1076



        matches := strset.NewWithSize(len(dnsIPs))




×







1077



        for _, ip := range dnsIPs {




×







1078



                matches.Add(fmt.Sprintf("ip%d.src == $%s && ip%d.dst == %s", af, pgAs, af, ip))




×







1079



        }




×







1080












1081



        policies, err := c.OVNNbClient.GetLogicalRouterPoliciesByExtID(c.config.ClusterRouter, "node", nodeName)




×







1082



        if err != nil {




×







1083



                klog.Errorf("failed to list logical router policies with external-ids:node = %q: %v", nodeName, err)




×







1084



                return err




×







1085



        }




×







1086












1087



        for _, policy := range policies {




×







1088



                if len(policy.ExternalIDs) == 0 || policy.ExternalIDs["vendor"] != util.CniTypeName || policy.ExternalIDs["isLocalDnsCache"] != "true" {




×







1089



                        continue




×







1090



                }










1091



                if policy.Priority == util.NodeRouterPolicyPriority && policy.Action == string(action) && slices.Equal(policy.Nexthops, nextHops) && matches.Has(policy.Match) {




×







1092



                        matches.Remove(policy.Match)




×







1093



                        continue




×







1094



                }










1095



                // delete unused policy router policy










1096



                klog.Infof("deleting logical router policy by UUID %s", policy.UUID)




×







1097



                if err = c.OVNNbClient.DeleteLogicalRouterPolicyByUUID(c.config.ClusterRouter, policy.UUID); err != nil {




×







1098



                        klog.Errorf("failed to delete logical router policy by UUID %s: %v", policy.UUID, err)




×







1099



                        return err




×







1100



                }




×







1101



        }










1102












1103



        for _, match := range matches.List() {




×







1104



                klog.Infof("add node local dns cache policy route for router %s: match %q, action %q, nexthop %q, externalID %v", c.config.ClusterRouter, match, action, nodeIP, externalIDs)




×







1105



                if err := c.addPolicyRouteToVpc(




×







1106



                        c.config.ClusterRouter,




×







1107



                        &kubeovnv1.PolicyRoute{




×







1108



                                Priority:  util.NodeRouterPolicyPriority,




×







1109



                                Match:     match,




×







1110



                                Action:    action,




×







1111



                                NextHopIP: nodeIP,




×







1112



                        },




×







1113



                        externalIDs,




×







1114



                ); err != nil {




×







1115



                        klog.Errorf("failed to add logical router policy for node %s: %v", nodeName, err)




×







1116



                        return err




×







1117



                }




×







1118



        }










1119












1120



        return nil




×







1121



}










1122












1123



func (c *Controller) deletePolicyRouteForLocalDNSCacheOnNode(nodeName string, af int) error {




×







1124



        policies, err := c.OVNNbClient.ListLogicalRouterPolicies(c.config.ClusterRouter, -1, map[string]string{




×







1125



                "vendor":          util.CniTypeName,




×







1126



                "node":            nodeName,




×







1127



                "address-family":  strconv.Itoa(af),




×







1128



                "isLocalDnsCache": "true",




×







1129



        }, true)




×







1130



        if err != nil {




×







1131



                klog.Errorf("failed to list logical router policies: %v", err)




×







1132



                return err




×







1133



        }




×







1134



        if len(policies) == 0 {




×







1135



                return nil




×







1136



        }




×







1137












1138



        for _, policy := range policies {




×







1139



                klog.Infof("delete node local dns cache policy route for router %s with match %s", c.config.ClusterRouter, policy.Match)




×







1140






×







1141



                if err := c.OVNNbClient.DeleteLogicalRouterPolicyByUUID(c.config.ClusterRouter, policy.UUID); err != nil {




×







1142



                        klog.Errorf("failed to delete policy route for node local dns in router %s with match %s: %v", c.config.ClusterRouter, policy.Match, err)




×







1143



                        return err




×







1144



                }




×







1145



        }










1146



        return nil




×







1147



}
























    

  • 

    •