kubeovn / kube-ovn / 17227113524

coverage: 21.226% (-0.2%) from 21.39%
17227113524

push

github

web-flow 
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.

10589 of 49886 relevant lines covered (21.23%)

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, err := util.IsNodeExcludedFromProviderNetwork(node, pn)




×







301



                if err != nil {




×







302



                        klog.Error(err)




×







303



                        return err




×







304



                }




×







305












306



                // Handle node annotation for exclusion (only when nodeSelector is not set)










307



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




×







308



                        newPn = pn.DeepCopy()




×







309



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




×







310



                        excluded = true




×







311



                }




×







312












313



                var customInterface string




×







314



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




×







315



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




×







316



                                customInterface = v.Interface




×







317



                                break




×







318



                        }










319



                }










320



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




×







321



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




×







322



                                if newPn == nil {




×







323



                                        newPn = pn.DeepCopy()




×







324



                                }




×







325



                                var index *int




×







326



                                for i := range newPn.Spec.CustomInterfaces {




×







327



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




×







328



                                                index = &i




×







329



                                                break




×







330



                                        }










331



                                }










332



                                if index != nil {




×







333



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




×







334



                                } else {




×







335



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




×







336



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




×







337



                                }




×







338



                        }










339



                }










340












341



                if newPn != nil {




×







342



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




×







343



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




×







344



                                return err




×







345



                        }




×







346



                }










347












348



                if len(node.Annotations) != 0 {




×







349



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




×







350



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




×







351



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




×







352



                                return err




×







353



                        }




×







354



                }










355












356



                if excluded {




×







357



                        if newPn == nil {




×







358



                                newPn = pn.DeepCopy()




×







359



                        } else {




×







360



                                newPn = newPn.DeepCopy()




×







361



                        }




×







362












363



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




×







364



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




×







365



                                if err != nil {




×







366



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




×







367



                                        return err




×







368



                                }




×







369



                        }










370



                }










371



        }










372












373



        return nil




×







374



}










375












376



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




×







377



        c.nodeKeyMutex.LockKey(key)




×







378



        defer func() {




×







379



                _ = c.nodeKeyMutex.UnlockKey(key)




×







380



                if err == nil {




×







381



                        c.deletingNodeObjMap.Delete(key)




×







382



                }




×







383



        }()










384



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




×







385






×







386



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




×







387



        if !ok {




×







388



                return nil




×







389



        }




×







390



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




×







391



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




×







392



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




×







393



                return nil




×







394



        }




×







395



        return c.deleteNode(key)




×







396



}










397












398



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




×







399



        portName := util.NodeLspName(key)




×







400



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




×







401



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




×







402



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




×







403



                return err




×







404



        }




×







405



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




×







406



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




×







407



                return err




×







408



        }




×







409












410



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




×







411



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




×







412



                        klog.Error(err)




×







413



                        return err




×







414



                }




×







415



        }










416












417



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










418



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




×







419



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




×







420



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




×







421



                return err




×







422



        }




×







423












424



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




×







425



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




×







426



                return err




×







427



        }




×







428












429



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




×







430



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




×







431



                return err




×







432



        }




×







433



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




×







434



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




×







435



                return err




×







436



        }




×







437












438



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




×







439



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




×







440






×







441



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




×







442



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




×







443



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




×







444



                return err




×







445



        }




×







446












447



        for _, pn := range providerNetworks {




×







448



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




×







449



                        klog.Error(err)




×







450



                        return err




×







451



                }




×







452



        }










453



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




×







454



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




×







455



                return err




×







456



        }




×







457












458



        return nil




×







459



}










460












461



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




×







462



        // update provider network status




×







463



        var needUpdate bool




×







464



        newPn := pn.DeepCopy()




×







465



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




×







466



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




×







467



                needUpdate = true




×







468



        }




×







469



        if newPn.Status.RemoveNodeConditions(node) {




×







470



                needUpdate = true




×







471



        }




×







472



        if needUpdate {




×







473



                var err error




×







474



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




×







475



                if err != nil {




×







476



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




×







477



                        return err




×







478



                }




×







479



        }










480












481



        // update provider network spec










482



        pn, newPn = newPn, nil




×







483



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




×







484



                newPn = pn.DeepCopy()




×







485



                newPn.Spec.ExcludeNodes = excludeNodes




×







486



        }




×







487












488



        var changed bool




×







489



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




×







490



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




×







491



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




×







492



                if !changed {




×







493



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




×







494



                }




×







495



                if len(nodes) != 0 {




×







496



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




×







497



                }




×







498



        }










499



        if changed {




×







500



                newPn = pn.DeepCopy()




×







501



                newPn.Spec.CustomInterfaces = customInterfaces




×







502



        }




×







503



        if newPn != nil {




×







504



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




×







505



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




×







506



                        return err




×







507



                }




×







508



        }










509












510



        return nil




×







511



}










512












513



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




×







514



        c.nodeKeyMutex.LockKey(key)




×







515



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




×







516



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




×







517






×







518



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




×







519



        if err != nil {




×







520



                if k8serrors.IsNotFound(err) {




×







521



                        return nil




×







522



                }




×







523



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




×







524



                return err




×







525



        }










526












527



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




×







528



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




×







529



                return err




×







530



        }




×







531












532



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




×







533



        if err != nil {




×







534



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




×







535



                return err




×







536



        }




×







537












538



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




×







539



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




×







540



                return err




×







541



        }




×







542



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




×







543



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




×







544



                return err




×







545



        }




×







546












547



        for _, cachedSubnet := range subnets {




×







548



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




×







549



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




×







550



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




×







551



                        c.addOrUpdateSubnetQueue.Add(cachedSubnet.Name)




×







552



                        continue




×







553



                }










554



                subnet := cachedSubnet.DeepCopy()




×







555



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




×







556



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




×







557



                                klog.Error(err)




×







558



                                return err




×







559



                        }




×







560



                }










561



        }










562












563



        return nil




×







564



}










565












566



func (c *Controller) checkSubnetGateway() {




×







567



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




×







568



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




×







569



        }




×







570



}










571












572



func (c *Controller) checkSubnetGatewayNode() error {




×







573



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




×







574



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




×







575



        if err != nil {




×







576



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




×







577



                return err




×







578



        }




×







579



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




×







580



        if err != nil {




×







581



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




×







582



                return err




×







583



        }




×







584












585



        for _, subnet := range subnetList {




×







586



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




×







587



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




×







588



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




×







589



                        subnet.Spec.GatewayNode == "" ||




×







590



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




×







591



                        !subnet.Spec.EnableEcmp {




×







592



                        continue




×







593



                }










594



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




×







595



                if len(gwNodes) < 2 {




×







596



                        continue




×







597



                }










598












599



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




×







600



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




×







601



                        if err != nil {




×







602



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




×







603



                                continue




×







604



                        }










605



                        for _, node := range nodes {




×







606



                                ipStr := node.Annotations[util.IPAddressAnnotation]




×







607



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




×







608



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




×







609



                                                continue




×







610



                                        }










611












612



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




×







613



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




×







614



                                                pinger, err := goping.NewPinger(ip)




×







615



                                                if err != nil {




×







616



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




×







617



                                                }




×







618



                                                pinger.SetPrivileged(true)




×







619






×







620



                                                count := 5




×







621



                                                pinger.Count = count




×







622



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




×







623



                                                pinger.Interval = 1 * time.Second




×







624






×







625



                                                var pingSucceeded bool




×







626



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




×







627



                                                        pingSucceeded = true




×







628



                                                        pinger.Stop()




×







629



                                                }




×







630



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




×







631



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




×







632



                                                        return err




×







633



                                                }




×







634












635



                                                nodeIsReady := nodeReady(node)




×







636



                                                if !pingSucceeded || !nodeIsReady {




×







637



                                                        if exist {




×







638



                                                                if !pingSucceeded {




×







639



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




×







640



                                                                }




×







641



                                                                if !nodeIsReady {




×







642



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




×







643



                                                                }




×







644



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




×







645



                                                                nextHops.Remove(ip)




×







646



                                                                delete(nameIPMap, node.Name)




×







647



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




×







648



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




×







649



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




×







650



                                                                        return err




×







651



                                                                }




×







652



                                                        }










653



                                                } else {




×







654



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




×







655



                                                        if !exist {




×







656



                                                                nextHops.Add(ip)




×







657



                                                                if nameIPMap == nil {




×







658



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




×







659



                                                                }




×







660



                                                                nameIPMap[node.Name] = ip




×







661



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




×







662



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




×







663



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




×







664



                                                                        return err




×







665



                                                                }




×







666



                                                        }










667



                                                }










668



                                        } else if exist {




×







669



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




×







670



                                                nextHops.Remove(ip)




×







671



                                                delete(nameIPMap, node.Name)




×







672



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




×







673



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




×







674



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




×







675



                                                        return err




×







676



                                                }




×







677



                                        }










678



                                }










679



                        }










680



                }










681



        }










682



        return nil




×







683



}










684












685



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




×







686



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




×







687



        var err error




×







688



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




×







689



                return nil




×







690



        }




×







691



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




×







692



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




×







693



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




×







694



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




×







695



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




×







696



                        return err




×







697



                }




×







698



        }










699



        return err




×







700



}










701












702



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




×







703



        i := 0




×







704



        for ; ; i++ {




×







705



                err = f(node)




×







706



                if err == nil {




×







707



                        return




×







708



                }




×







709



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




×







710



                if i >= (attempts - 1) {




×







711



                        break




×







712



                }










713



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




×







714



        }










715



        if i >= (attempts - 1) {




×







716



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




×







717



                klog.Error(errMsg)




×







718



                return errMsg




×







719



        }




×







720



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




×







721



        return nil




×







722



}










723












724



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




×







725



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




×







726



        for _, pod := range pods {




×







727



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




×







728



                        continue




×







729



                }










730












731



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




×







732



                        subnetName := pod.Annotations[util.LogicalSwitchAnnotation]




×







733



                        if subnetName == "" {




×







734



                                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)




×







735



                                continue




×







736



                        }










737












738



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




×







739



                        if err != nil {




×







740



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




×







741



                                return nil, err




×







742



                        }




×







743












744



                        if subnet.Spec.Vlan == "" {




×







745



                                continue




×







746



                        }










747



                }










748












749



                podName := c.getNameByPod(pod)




×







750






×







751



                podNets, err := c.getPodKubeovnNets(pod)




×







752



                if err != nil {




×







753



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




×







754



                        return nil, err




×







755



                }




×







756












757



                for _, podNet := range podNets {




×







758



                        if !isOvnSubnet(podNet.Subnet) {




×







759



                                continue




×







760



                        }










761












762



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




×







763



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




×







764



                        }




×







765



                }










766



        }










767



        return ports, nil




×







768



}










769












770



func (c *Controller) CheckNodePortGroup() {




×







771



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




×







772



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




×







773



        }




×







774



}










775












776



func (c *Controller) checkAndUpdateNodePortGroup() error {




×







777



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




×







778



        var networkPolicyExists bool




×







779



        if c.config.EnableNP {




×







780



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




×







781



                networkPolicyExists = len(np) != 0




×







782



        }




×







783












784



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




×







785



        if err != nil {




×







786



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




×







787



                return err




×







788



        }




×







789












790



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




×







791



        if err != nil {




×







792



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




×







793



                return err




×







794



        }




×







795












796



        for _, node := range nodes {




×







797



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




×







798



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




×







799






×







800



                // use join IP only when no internal IP exists




×







801



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




×







802



                joinIP := node.Annotations[util.IPAddressAnnotation]




×







803



                joinIPv4, joinIPv6 := util.SplitStringIP(joinIP)




×







804



                if nodeIPv4 == "" {




×







805



                        nodeIPv4 = joinIPv4




×







806



                }




×







807



                if nodeIPv6 == "" {




×







808



                        nodeIPv6 = joinIPv6




×







809



                }




×







810



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




×







811






×







812



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




×







813



                if err != nil {




×







814



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




×







815



                        return err




×







816



                }




×







817












818



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




×







819



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




×







820



                        return err




×







821



                }




×







822












823



                if networkPolicyExists {




×







824



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




×







825



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




×







826



                        }




×







827



                } else {




×







828



                        // clear all acl




×







829



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




×







830



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




×







831



                        }




×







832



                }










833



        }










834












835



        return nil




×







836



}










837












838



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




×







839



        annoChassisName := node.Annotations[util.ChassisAnnotation]




×







840



        if annoChassisName == "" {




×







841



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




×







842



                return nil




×







843



        }




×







844



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




×







845



        if err != nil {




×







846



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




×







847



                return err




×







848



        }




×







849



        if chassis == nil {




×







850



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




×







851



                // chassis name conflict, do GC




×







852



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




×







853



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




×







854



                        return err




×







855



                }




×







856



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




×







857



                klog.Error(err)




×







858



                return err




×







859



        }










860












861



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




×







862



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




×







863



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




×







864



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




×







865



                        klog.Error(err)




×







866



                        return err




×







867



                }




×







868



        }










869



        return nil




×







870



}










871












872



func (c *Controller) addNodeGatewayStaticRoute() error {




×







873



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




×







874



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




×







875



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




×







876



                if err != nil {




×







877



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




×







878



                }




×







879



                if len(existRoute) != 0 {




×







880



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




×







881



                        return nil




×







882



                }




×







883



        }










884



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




×







885



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




×







886



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




×







887



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




×







888



                                continue




×







889



                        }










890












891



                        if err := c.addStaticRouteToVpc(




×







892



                                c.config.ClusterRouter,




×







893



                                &kubeovnv1.StaticRoute{




×







894



                                        Policy:     kubeovnv1.PolicyDst,




×







895



                                        CIDR:       cidrBlock,




×







896



                                        NextHopIP:  nextHop,




×







897



                                        RouteTable: util.MainRouteTable,




×







898



                                },




×







899



                        ); err != nil {




×







900



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




×







901



                                return err




×







902



                        }




×







903



                }










904



        }










905



        return nil




×







906



}










907












908



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




×







909



        ipSuffix := "ip4"




×







910



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




×







911



                ipSuffix = "ip6"




×







912



        }




×







913



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




×







914



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




×







915



        if err != nil {




×







916



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




×







917



                return nil, nil, err




×







918



        }




×







919



        if len(policyList) == 0 {




×







920



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




×







921



        }




×







922



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




×







923



}










924












925



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




×







926



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




×







927



        if err != nil {




×







928



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




×







929



                return err




×







930



        }




×







931












932



        addresses := c.ipam.GetPodAddress(portName)




×







933



        for _, addr := range addresses {




×







934



                if addr.IP == "" {




×







935



                        continue




×







936



                }










937



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




×







938



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




×







939



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




×







940



                        return err




×







941



                }




×







942



        }










943












944



        for _, subnet := range subnets {




×







945



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




×







946



                        continue




×







947



                }










948












949



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




×







950



                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, nodeName)




×







951



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




×







952



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




×







953



                                return err




×







954



                        }




×







955












956



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




×







957



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




×







958



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




×







959



                                return err




×







960



                        }




×







961



                }










962












963



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




×







964



                        if subnet.Spec.EnableEcmp {




×







965



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




×







966



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




×







967



                                        if err != nil {




×







968



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




×







969



                                                continue




×







970



                                        }










971












972



                                        exist := false




×







973



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




×







974



                                                exist = true




×







975



                                        }




×







976












977



                                        if exist {




×







978



                                                nextHops.Remove(nameIPMap[nodeName])




×







979



                                                delete(nameIPMap, nodeName)




×







980






×







981



                                                if nextHops.Size() == 0 {




×







982



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




×







983



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




×







984



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




×







985



                                                                return err




×







986



                                                        }




×







987



                                                } else {




×







988



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




×







989



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




×







990



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




×







991



                                                                return err




×







992



                                                        }




×







993



                                                }










994



                                        }










995



                                }










996



                        } else {




×







997



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




×







998



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




×







999



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




×







1000



                                        return err




×







1001



                                }




×







1002



                        }










1003



                }










1004



        }










1005



        return nil




×







1006



}










1007












1008



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




×







1009



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




×







1010



        if err != nil {




×







1011



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




×







1012



                return err




×







1013



        }




×







1014












1015



        for _, subnet := range subnets {




×







1016



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




×







1017



                        continue




×







1018



                }










1019












1020



                if subnet.Spec.EnableEcmp {




×







1021



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




×







1022



                                continue




×







1023



                        }










1024












1025



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




×







1026



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




×







1027



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




×







1028



                                                continue




×







1029



                                        }










1030












1031



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




×







1032



                                        if err != nil {




×







1033



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




×







1034



                                                continue




×







1035



                                        }










1036



                                        if nameIPMap[nodeName] == nextHop {




×







1037



                                                continue




×







1038



                                        }










1039












1040



                                        nextHops.Add(nextHop)




×







1041



                                        if nameIPMap == nil {




×







1042



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




×







1043



                                        }




×







1044



                                        nameIPMap[nodeName] = nextHop




×







1045



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




×







1046



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




×







1047



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




×







1048



                                                return err




×







1049



                                        }




×







1050



                                }










1051



                        }










1052



                } else {




×







1053



                        if subnet.Status.ActivateGateway != nodeName {




×







1054



                                continue




×







1055



                        }










1056



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




×







1057



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




×







1058



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




×







1059



                                return err




×







1060



                        }




×







1061



                }










1062



        }










1063



        return nil




×







1064



}










1065












1066



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




×







1067



        if len(dnsIPs) == 0 {




×







1068



                return c.deletePolicyRouteForLocalDNSCacheOnNode(nodeName, af)




×







1069



        }




×







1070












1071



        var (




×







1072



                externalIDs = map[string]string{




×







1073



                        "vendor":          util.CniTypeName,




×







1074



                        "node":            nodeName,




×







1075



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




×







1076



                        "isLocalDnsCache": "true",




×







1077



                }




×







1078



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




×







1079



                action   = kubeovnv1.PolicyRouteActionReroute




×







1080



                nextHops = []string{nodeIP}




×







1081



        )




×







1082



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




×







1083



        for _, ip := range dnsIPs {




×







1084



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




×







1085



        }




×







1086












1087



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




×







1088



        if err != nil {




×







1089



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




×







1090



                return err




×







1091



        }




×







1092












1093



        for _, policy := range policies {




×







1094



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




×







1095



                        continue




×







1096



                }










1097



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




×







1098



                        matches.Remove(policy.Match)




×







1099



                        continue




×







1100



                }










1101



                // delete unused policy router policy










1102



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




×







1103



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




×







1104



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




×







1105



                        return err




×







1106



                }




×







1107



        }










1108












1109



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




×







1110



                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)




×







1111



                if err := c.addPolicyRouteToVpc(




×







1112



                        c.config.ClusterRouter,




×







1113



                        &kubeovnv1.PolicyRoute{




×







1114



                                Priority:  util.NodeRouterPolicyPriority,




×







1115



                                Match:     match,




×







1116



                                Action:    action,




×







1117



                                NextHopIP: nodeIP,




×







1118



                        },




×







1119



                        externalIDs,




×







1120



                ); err != nil {




×







1121



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




×







1122



                        return err




×







1123



                }




×







1124



        }










1125












1126



        return nil




×







1127



}










1128












1129



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




×







1130



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




×







1131



                "vendor":          util.CniTypeName,




×







1132



                "node":            nodeName,




×







1133



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




×







1134



                "isLocalDnsCache": "true",




×







1135



        }, true)




×







1136



        if err != nil {




×







1137



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




×







1138



                return err




×







1139



        }




×







1140



        if len(policies) == 0 {




×







1141



                return nil




×







1142



        }




×







1143












1144



        for _, policy := range policies {




×







1145



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




×







1146






×







1147



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




×







1148



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




×







1149



                        return err




×







1150



                }




×







1151



        }










1152



        return nil




×







1153



}
























    

  • 

    •