kubeovn / kube-ovn / 19809726130

coverage: 22.301% (-0.01%) from 22.312%
19809726130

push

github

web-flow 
feat: add gatewayNodeSelectors (#5956)

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

14 of 141 new or added lines in 7 files covered. (9.93%)

3 existing lines in 3 files now uncovered.

11548 of 51782 relevant lines covered (22.3%)

0.26 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) ||




×








NEW


61



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




×








NEW


62



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




×







63



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




×







64



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




×







65



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




×







66



                        c.addNodeQueue.Add(key)




×







67



                } else {




×







68



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




×







69



                        c.updateNodeQueue.Add(key)




×







70



                }




×







71



        }










72



}










73












74



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




×







75



        var node *v1.Node




×







76



        switch t := obj.(type) {




×







77



        case *v1.Node:




×







78



                node = t




×







79



        case cache.DeletedFinalStateUnknown:




×







80



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




×







81



                if !ok {




×







82



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




×







83



                        return




×







84



                }




×







85



                node = n




×







86



        default:




×







87



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




×







88



                return




×







89



        }










90












91



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




×







92



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




×







93



        c.deletingNodeObjMap.Store(key, node)




×







94



        c.deleteNodeQueue.Add(key)




×







95



}










96












97



func nodeUnderlayAddressSetName(node string, af int) string {




×







98



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




×







99



}




×







100












101



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




×







102



        c.nodeKeyMutex.LockKey(key)




×







103



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




×







104












105



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




×







106



        if err != nil {




×







107



                if k8serrors.IsNotFound(err) {




×







108



                        return nil




×







109



                }




×







110



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




×







111



                return err




×







112



        }










113



        node := cachedNode.DeepCopy()




×







114



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




×







115






×







116



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




×







117



        if err != nil {




×







118



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




×







119



                return err




×







120



        }




×







121












122



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




×







123



        for _, subnet := range subnets {




×







124



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




×







125



                        continue




×







126



                }










127












128



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




×







129



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




×







130



                        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)




×







131



                        klog.Warning(msg)




×







132



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




×







133



                        break




×







134



                }










135



        }










136












137



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




×







138



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




×







139



                return err




×







140



        }




×







141












142



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




×







143



        if err != nil {




×







144



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




×







145



                return err




×







146



        }




×







147












148



        var v4IP, v6IP, mac string




×







149






×







150



        portName := util.NodeLspName(key)




×







151



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




×







152



                macStr := node.Annotations[util.MacAddressAnnotation]




×







153



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




×







154



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




×







155



                if err != nil {




×







156



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




×







157



                        return err




×







158



                }




×







159



        } else {




×







160



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




×







161



                if err != nil {




×







162



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




×







163



                        return err




×







164



                }




×







165












166



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










167



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




×







168



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




×







169



                        return err




×







170



                }




×







171



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




×







172



        }










173












174



        ipStr := util.GetStringIP(v4IP, v6IP)




×







175



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




×







176



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




×







177



                return err




×







178



        }




×







179












180



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




×







181



                if ip == "" {




×







182



                        continue




×







183



                }










184












185



                nodeIP, af := nodeIPv4, 4




×







186



                protocol := util.CheckProtocol(ip)




×







187



                if protocol == kubeovnv1.ProtocolIPv6 {




×







188



                        nodeIP, af = nodeIPv6, 6




×







189



                }




×







190



                if nodeIP != "" {




×







191



                        var (




×







192



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




×







193



                                action      = kubeovnv1.PolicyRouteActionReroute




×







194



                                externalIDs = map[string]string{




×







195



                                        "vendor":         util.CniTypeName,




×







196



                                        "node":           node.Name,




×







197



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




×







198



                                }




×







199



                        )




×







200



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




×







201



                        if err = c.addPolicyRouteToVpc(




×







202



                                c.config.ClusterRouter,




×







203



                                &kubeovnv1.PolicyRoute{




×







204



                                        Priority:  util.NodeRouterPolicyPriority,




×







205



                                        Match:     match,




×







206



                                        Action:    action,




×







207



                                        NextHopIP: ip,




×







208



                                },




×







209



                                externalIDs,




×







210



                        ); err != nil {




×







211



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




×







212



                                return err




×







213



                        }




×







214












215



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




×







216



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




×







217



                                if util.CheckProtocol(ip) == protocol {




×







218



                                        dnsIPs = append(dnsIPs, ip)




×







219



                                }




×







220



                        }










221












222



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




×







223



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




×







224



                                return err




×







225



                        }




×







226



                }










227



        }










228












229



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




×







230



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




×







231



                return err




×







232



        }




×







233












234



        patch := util.KVPatch{




×







235



                util.IPAddressAnnotation:     ipStr,




×







236



                util.MacAddressAnnotation:    mac,




×







237



                util.CidrAnnotation:          subnet.Spec.CIDRBlock,




×







238



                util.GatewayAnnotation:       subnet.Spec.Gateway,




×







239



                util.LogicalSwitchAnnotation: c.config.NodeSwitch,




×







240



                util.AllocatedAnnotation:     "true",




×







241



                util.PortNameAnnotation:      portName,




×







242



        }




×







243



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




×







244



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




×







245



                return err




×







246



        }




×







247












248



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




×







249



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




×







250



                return err




×







251



        }




×







252












253



        for _, subnet := range subnets {




×







254



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




×







255



                        continue




×







256



                }










257



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




×







258



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




×







259



                        return err




×







260



                }




×







261



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










262



                c.addOrUpdateSubnetQueue.Add(subnet.Name)




×







263



        }










264












265



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










266



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




×







267



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




×







268



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




×







269



                return err




×







270



        }




×







271













NEW


272



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




×







273



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




×







274



                return err




×







275



        }




×







276












277



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




×







278



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




×







279



                return err




×







280



        }




×







281












282



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




×







283



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




×







284



                return err




×







285



        }




×







286



        return nil




×







287



}










288












289



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




×







290



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




×







291



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




×







292



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




×







293



                return err




×







294



        }




×







295












296



        for _, pn := range providerNetworks {




×







297



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




×







298



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




×







299






×







300



                var newPn *kubeovnv1.ProviderNetwork




×







301



                excluded, err := util.IsNodeExcludedFromProviderNetwork(node, pn)




×







302



                if err != nil {




×







303



                        klog.Error(err)




×







304



                        return err




×







305



                }




×







306












307



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










308



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




×







309



                        newPn = pn.DeepCopy()




×







310



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




×







311



                        excluded = true




×







312



                }




×







313












314



                var customInterface string




×







315



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




×







316



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




×







317



                                customInterface = v.Interface




×







318



                                break




×







319



                        }










320



                }










321



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




×







322



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




×







323



                                if newPn == nil {




×







324



                                        newPn = pn.DeepCopy()




×







325



                                }




×







326



                                var index *int




×







327



                                for i := range newPn.Spec.CustomInterfaces {




×







328



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




×







329



                                                index = &i




×







330



                                                break




×







331



                                        }










332



                                }










333



                                if index != nil {




×







334



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




×







335



                                } else {




×







336



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




×







337



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




×







338



                                }




×







339



                        }










340



                }










341












342



                if newPn != nil {




×







343



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




×







344



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




×







345



                                return err




×







346



                        }




×







347



                }










348












349



                if len(node.Annotations) != 0 {




×







350



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




×







351



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




×







352



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




×







353



                                return err




×







354



                        }




×







355



                }










356












357



                if excluded {




×







358



                        if newPn == nil {




×







359



                                newPn = pn.DeepCopy()




×







360



                        } else {




×







361



                                newPn = newPn.DeepCopy()




×







362



                        }




×







363












364



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




×







365



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




×







366



                                if err != nil {




×







367



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




×







368



                                        return err




×







369



                                }




×







370



                        }










371



                }










372



        }










373












374



        return nil




×







375



}










376












377



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




×







378



        c.nodeKeyMutex.LockKey(key)




×







379



        defer func() {




×







380



                _ = c.nodeKeyMutex.UnlockKey(key)




×







381



                if err == nil {




×







382



                        c.deletingNodeObjMap.Delete(key)




×







383



                }




×







384



        }()










385



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




×







386






×







387



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




×







388



        if !ok {




×







389



                return nil




×







390



        }




×







391



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




×







392



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




×







393



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




×







394



                return nil




×







395



        }




×







396



        return c.deleteNode(key)




×







397



}










398












399



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




×







400



        portName := util.NodeLspName(key)




×







401



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




×







402



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




×







403



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




×







404



                return err




×







405



        }




×







406



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




×







407



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




×







408



                return err




×







409



        }




×







410












411



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




×







412



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




×







413



                        klog.Error(err)




×







414



                        return err




×







415



                }




×







416



        }










417












418



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










419



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




×







420



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




×







421



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




×







422



                return err




×







423



        }




×







424












425



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




×







426



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




×







427



                return err




×







428



        }




×







429












430



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




×







431



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




×







432



                return err




×







433



        }




×







434



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




×







435



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




×







436



                return err




×







437



        }




×







438












439



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




×







440



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




×







441






×







442



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




×







443



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




×







444



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




×







445



                return err




×







446



        }




×







447












448



        for _, pn := range providerNetworks {




×







449



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




×







450



                        klog.Error(err)




×







451



                        return err




×







452



                }




×







453



        }










454



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




×







455



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




×







456



                return err




×







457



        }




×







458












459



        return nil




×







460



}










461












462



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




×







463



        // update provider network status




×







464



        var needUpdate bool




×







465



        newPn := pn.DeepCopy()




×







466



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




×







467



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




×







468



                needUpdate = true




×







469



        }




×







470



        if newPn.Status.RemoveNodeConditions(node) {




×







471



                needUpdate = true




×







472



        }




×







473



        if needUpdate {




×







474



                var err error




×







475



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




×







476



                if err != nil {




×







477



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




×







478



                        return err




×







479



                }




×







480



        }










481












482



        // update provider network spec










483



        pn, newPn = newPn, nil




×







484



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




×







485



                newPn = pn.DeepCopy()




×







486



                newPn.Spec.ExcludeNodes = excludeNodes




×







487



        }




×







488












489



        var changed bool




×







490



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




×







491



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




×







492



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




×







493



                if !changed {




×







494



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




×







495



                }




×







496



                if len(nodes) != 0 {




×







497



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




×







498



                }




×







499



        }










500



        if changed {




×







501



                newPn = pn.DeepCopy()




×







502



                newPn.Spec.CustomInterfaces = customInterfaces




×







503



        }




×







504



        if newPn != nil {




×







505



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




×







506



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




×







507



                        return err




×







508



                }




×







509



        }










510












511



        return nil




×







512



}










513












514



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




×







515



        c.nodeKeyMutex.LockKey(key)




×







516



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




×







517



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




×







518






×







519



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




×







520



        if err != nil {




×







521



                if k8serrors.IsNotFound(err) {




×







522



                        return nil




×







523



                }




×







524



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




×







525



                return err




×







526



        }










527












528



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




×







529



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




×







530



                return err




×







531



        }




×







532












533



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




×







534



        if err != nil {




×







535



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




×







536



                return err




×







537



        }




×







538












539



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




×







540



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




×







541



                return err




×







542



        }




×







543



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




×







544



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




×







545



                return err




×







546



        }




×







547












548



        for _, cachedSubnet := range subnets {




×







549



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




×







550



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




×







551



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




×







552



                        c.addOrUpdateSubnetQueue.Add(cachedSubnet.Name)




×







553



                        continue




×







554



                }










555



                subnet := cachedSubnet.DeepCopy()




×








NEW


556






×








NEW


557



                // For subnets using GatewayNodeSelectors, always trigger reconciliation




×








NEW


558



                // when node labels change, since the node might have been added or removed




×








NEW


559



                // from the gateway list




×








NEW


560



                if subnet.Spec.GatewayNode == "" && len(subnet.Spec.GatewayNodeSelectors) > 0 {




×








NEW


561



                        c.addOrUpdateSubnetQueue.Add(subnet.Name)




×








NEW


562



                        continue




×







563



                }










564












565



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




×







566



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




×







567



                                klog.Error(err)




×







568



                                return err




×







569



                        }




×







570



                }










571



        }










572












573



        return nil




×







574



}










575












576



func (c *Controller) checkSubnetGateway() {




×







577



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




×







578



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




×







579



        }




×







580



}










581












582



func (c *Controller) checkSubnetGatewayNode() error {




×







583



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




×







584



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




×







585



        if err != nil {




×







586



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




×







587



                return err




×







588



        }




×







589



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




×







590



        if err != nil {




×







591



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




×







592



                return err




×







593



        }




×







594












595



        for _, subnet := range subnetList {




×







596



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




×







597



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




×







598



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




×







599



                        subnet.Spec.GatewayNode == "" ||




×







600



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




×







601



                        !subnet.Spec.EnableEcmp {




×







602



                        continue




×







603



                }










604



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




×







605



                if len(gwNodes) < 2 {




×







606



                        continue




×







607



                }










608












609



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




×







610



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




×







611



                        if err != nil {




×







612



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




×







613



                                continue




×







614



                        }










615



                        for _, node := range nodes {




×







616



                                ipStr := node.Annotations[util.IPAddressAnnotation]




×







617



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




×







618



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




×







619



                                                continue




×







620



                                        }










621












622



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




×







623



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




×







624



                                                pinger, err := goping.NewPinger(ip)




×







625



                                                if err != nil {




×







626



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




×







627



                                                }




×







628



                                                pinger.SetPrivileged(true)




×







629






×







630



                                                count := 5




×







631



                                                pinger.Count = count




×







632



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




×







633



                                                pinger.Interval = 1 * time.Second




×







634






×







635



                                                var pingSucceeded bool




×







636



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




×







637



                                                        pingSucceeded = true




×







638



                                                        pinger.Stop()




×







639



                                                }




×







640



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




×







641



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




×







642



                                                        return err




×







643



                                                }




×







644












645



                                                nodeIsReady := nodeReady(node)




×







646



                                                if !pingSucceeded || !nodeIsReady {




×







647



                                                        if exist {




×







648



                                                                if !pingSucceeded {




×







649



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




×







650



                                                                }




×







651



                                                                if !nodeIsReady {




×







652



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




×







653



                                                                }




×







654



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




×







655



                                                                nextHops.Remove(ip)




×







656



                                                                delete(nameIPMap, node.Name)




×







657



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




×







658



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




×







659



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




×







660



                                                                        return err




×







661



                                                                }




×







662



                                                        }










663



                                                } else {




×







664



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




×







665



                                                        if !exist {




×







666



                                                                nextHops.Add(ip)




×







667



                                                                if nameIPMap == nil {




×







668



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




×







669



                                                                }




×







670



                                                                nameIPMap[node.Name] = ip




×







671



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




×







672



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




×







673



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




×







674



                                                                        return err




×







675



                                                                }




×







676



                                                        }










677



                                                }










678



                                        } else if exist {




×







679



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




×







680



                                                nextHops.Remove(ip)




×







681



                                                delete(nameIPMap, node.Name)




×







682



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




×







683



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




×







684



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




×







685



                                                        return err




×







686



                                                }




×







687



                                        }










688



                                }










689



                        }










690



                }










691



        }










692



        return nil




×







693



}










694












695



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




×







696



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




×







697



        var err error




×







698



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




×







699



                return nil




×







700



        }




×







701



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




×







702



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




×







703



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




×







704



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




×







705



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




×







706



                        return err




×







707



                }




×







708



        }










709



        return err




×







710



}










711












712



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




×







713



        i := 0




×







714



        for ; ; i++ {




×







715



                err = f(node)




×







716



                if err == nil {




×







717



                        return err




×







718



                }




×







719



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




×







720



                if i >= (attempts - 1) {




×







721



                        break




×







722



                }










723



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




×







724



        }










725



        if i >= (attempts - 1) {




×







726



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




×







727



                klog.Error(errMsg)




×







728



                return errMsg




×







729



        }




×







730



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




×







731



        return nil




×







732



}










733












734



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




×







735



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




×







736



        for _, pod := range pods {




×







737



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




×







738



                        continue




×







739



                }










740












741



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




×







742



                        subnetName := pod.Annotations[util.LogicalSwitchAnnotation]




×







743



                        if subnetName == "" {




×







744



                                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)




×







745



                                continue




×







746



                        }










747












748



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




×







749



                        if err != nil {




×







750



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




×







751



                                return nil, err




×







752



                        }




×







753












754



                        if subnet.Spec.Vlan == "" {




×







755



                                continue




×







756



                        }










757



                }










758












759



                podName := c.getNameByPod(pod)




×







760






×







761



                podNets, err := c.getPodKubeovnNets(pod)




×







762



                if err != nil {




×







763



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




×







764



                        return nil, err




×







765



                }




×







766












767



                for _, podNet := range podNets {




×







768



                        if !isOvnSubnet(podNet.Subnet) {




×







769



                                continue




×







770



                        }










771












772



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




×







773



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




×







774



                        }




×







775



                }










776



        }










777



        return ports, nil




×







778



}










779












780



func (c *Controller) CheckNodePortGroup() {




×







781



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




×







782



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




×







783



        }




×







784



}










785












786



func (c *Controller) checkAndUpdateNodePortGroup() error {




×







787



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




×







788



        var networkPolicyExists bool




×







789



        if c.config.EnableNP {




×







790



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




×







791



                networkPolicyExists = len(np) != 0




×







792



        }




×







793












794



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




×







795



        if err != nil {




×







796



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




×







797



                return err




×







798



        }




×







799












800



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




×







801



        if err != nil {




×







802



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




×







803



                return err




×







804



        }




×







805












806



        for _, node := range nodes {




×







807



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




×







808



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




×







809






×







810



                // use join IP only when no internal IP exists




×







811



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




×







812



                joinIP := node.Annotations[util.IPAddressAnnotation]




×







813



                joinIPv4, joinIPv6 := util.SplitStringIP(joinIP)




×







814



                if nodeIPv4 == "" {




×







815



                        nodeIPv4 = joinIPv4




×







816



                }




×







817



                if nodeIPv6 == "" {




×







818



                        nodeIPv6 = joinIPv6




×







819



                }




×







820



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




×







821






×







822



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




×







823



                if err != nil {




×







824



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




×







825



                        return err




×







826



                }




×







827












828



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




×







829



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




×







830



                        return err




×







831



                }




×







832












833



                if networkPolicyExists {




×







834



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




×







835



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




×







836



                        }




×







837



                } else {




×







838



                        // clear all acl




×







839



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




×







840



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




×







841



                        }




×







842



                }










843



        }










844












845



        return nil




×







846



}










847












848



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




×







849



        annoChassisName := node.Annotations[util.ChassisAnnotation]




×







850



        if annoChassisName == "" {




×







851



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




×







852



                return nil




×







853



        }




×







854



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




×







855



        if err != nil {




×







856



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




×







857



                return err




×







858



        }




×







859



        if chassis == nil {




×







860



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




×







861



                // chassis name conflict, do GC




×







862



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




×







863



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




×







864



                        return err




×







865



                }




×







866



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




×







867



                klog.Error(err)




×







868



                return err




×







869



        }










870












871



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




×







872



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




×







873



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




×







874



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




×







875



                        klog.Error(err)




×







876



                        return err




×







877



                }




×







878



        }










879



        return nil




×







880



}










881












882



func (c *Controller) addNodeGatewayStaticRoute() error {




×







883



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




×







884



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




×







885



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




×







886



                if err != nil {




×







887



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




×







888



                }




×







889



                if len(existRoute) != 0 {




×







890



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




×







891



                        return nil




×







892



                }




×







893



        }










894



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




×







895



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




×







896



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




×







897



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




×







898



                                continue




×







899



                        }










900












901



                        if err := c.addStaticRouteToVpc(




×







902



                                c.config.ClusterRouter,




×







903



                                &kubeovnv1.StaticRoute{




×







904



                                        Policy:     kubeovnv1.PolicyDst,




×







905



                                        CIDR:       cidrBlock,




×







906



                                        NextHopIP:  nextHop,




×







907



                                        RouteTable: util.MainRouteTable,




×







908



                                },




×







909



                        ); err != nil {




×







910



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




×







911



                                return err




×







912



                        }




×







913



                }










914



        }










915



        return nil




×







916



}










917












918



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




×







919



        ipSuffix := "ip4"




×







920



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




×







921



                ipSuffix = "ip6"




×







922



        }




×







923



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




×







924



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




×







925



        if err != nil {




×







926



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




×







927



                return nil, nil, err




×







928



        }




×







929



        if len(policyList) == 0 {




×







930



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




×







931



        }




×







932



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




×







933



}










934












935



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




×







936



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




×







937



        if err != nil {




×







938



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




×







939



                return err




×







940



        }




×







941












942



        addresses := c.ipam.GetPodAddress(portName)




×







943



        for _, addr := range addresses {




×







944



                if addr.IP == "" {




×







945



                        continue




×







946



                }










947



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




×







948



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




×







949



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




×







950



                        return err




×







951



                }




×







952



        }










953












954



        for _, subnet := range subnets {




×







955



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




×







956



                        continue




×







957



                }










958












959



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




×







960



                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, nodeName)




×







961



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




×







962



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




×







963



                                return err




×







964



                        }




×







965












966



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




×







967



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




×







968



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




×







969



                                return err




×







970



                        }




×







971



                }










972












973



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




×







974



                        if subnet.Spec.EnableEcmp {




×







975



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




×







976



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




×







977



                                        if err != nil {




×







978



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




×







979



                                                continue




×







980



                                        }










981












982



                                        exist := false




×







983



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




×







984



                                                exist = true




×







985



                                        }




×







986












987



                                        if exist {




×







988



                                                nextHops.Remove(nameIPMap[nodeName])




×







989



                                                delete(nameIPMap, nodeName)




×







990






×







991



                                                if nextHops.Size() == 0 {




×







992



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




×







993



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




×







994



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




×







995



                                                                return err




×







996



                                                        }




×







997



                                                } else {




×







998



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




×







999



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




×







1000



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




×







1001



                                                                return err




×







1002



                                                        }




×







1003



                                                }










1004



                                        }










1005



                                }










1006



                        } else {




×







1007



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




×







1008



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




×







1009



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




×







1010



                                        return err




×







1011



                                }




×







1012



                        }










1013



                }










1014



        }










1015



        return nil




×







1016



}










1017













NEW


1018



func (c *Controller) addPolicyRouteForCentralizedSubnetOnNode(node *v1.Node, nodeIP string) error {




×







1019



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




×







1020



        if err != nil {




×







1021



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




×







1022



                return err




×







1023



        }




×







1024












1025



        for _, subnet := range subnets {




×







1026



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




×







1027



                        continue




×







1028



                }











NEW


1029



                nodeName := node.Name




×







1030



                if subnet.Spec.EnableEcmp {




×








NEW


1031



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




×








NEW


1032



                                (subnet.Spec.GatewayNode != "" || !util.MatchLabelSelectors(subnet.Spec.GatewayNodeSelectors, node.Labels)) {




×








UNCOV

1033



                                continue




×







1034



                        }










1035












1036



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




×







1037



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




×







1038



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




×







1039



                                                continue




×







1040



                                        }










1041












1042



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




×







1043



                                        if err != nil {




×







1044



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




×







1045



                                                continue




×







1046



                                        }










1047



                                        if nameIPMap[nodeName] == nextHop {




×







1048



                                                continue




×







1049



                                        }










1050












1051



                                        nextHops.Add(nextHop)




×







1052



                                        if nameIPMap == nil {




×







1053



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




×







1054



                                        }




×







1055



                                        nameIPMap[nodeName] = nextHop




×







1056



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




×







1057



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




×







1058



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




×







1059



                                                return err




×







1060



                                        }




×







1061



                                }










1062



                        }










1063



                } else {




×







1064



                        if subnet.Status.ActivateGateway != nodeName {




×







1065



                                continue




×







1066



                        }










1067



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




×







1068



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




×







1069



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




×







1070



                                return err




×







1071



                        }




×







1072



                }










1073



        }










1074



        return nil




×







1075



}










1076












1077



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




×







1078



        if len(dnsIPs) == 0 {




×







1079



                return c.deletePolicyRouteForLocalDNSCacheOnNode(nodeName, af)




×







1080



        }




×







1081












1082



        var (




×







1083



                externalIDs = map[string]string{




×







1084



                        "vendor":          util.CniTypeName,




×







1085



                        "node":            nodeName,




×







1086



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




×







1087



                        "isLocalDnsCache": "true",




×







1088



                }




×







1089



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




×







1090



                action   = kubeovnv1.PolicyRouteActionReroute




×







1091



                nextHops = []string{nodeIP}




×







1092



        )




×







1093



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




×







1094



        for _, ip := range dnsIPs {




×







1095



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




×







1096



        }




×







1097












1098



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




×







1099



        if err != nil {




×







1100



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




×







1101



                return err




×







1102



        }




×







1103












1104



        for _, policy := range policies {




×







1105



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




×







1106



                        continue




×







1107



                }










1108



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




×







1109



                        matches.Remove(policy.Match)




×







1110



                        continue




×







1111



                }










1112



                // delete unused policy router policy










1113



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




×







1114



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




×







1115



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




×







1116



                        return err




×







1117



                }




×







1118



        }










1119












1120



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




×







1121



                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)




×







1122



                if err := c.addPolicyRouteToVpc(




×







1123



                        c.config.ClusterRouter,




×







1124



                        &kubeovnv1.PolicyRoute{




×







1125



                                Priority:  util.NodeRouterPolicyPriority,




×







1126



                                Match:     match,




×







1127



                                Action:    action,




×







1128



                                NextHopIP: nodeIP,




×







1129



                        },




×







1130



                        externalIDs,




×







1131



                ); err != nil {




×







1132



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




×







1133



                        return err




×







1134



                }




×







1135



        }










1136












1137



        return nil




×







1138



}










1139












1140



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




×







1141



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




×







1142



                "vendor":          util.CniTypeName,




×







1143



                "node":            nodeName,




×







1144



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




×







1145



                "isLocalDnsCache": "true",




×







1146



        }, true)




×







1147



        if err != nil {




×







1148



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




×







1149



                return err




×







1150



        }




×







1151



        if len(policies) == 0 {




×







1152



                return nil




×







1153



        }




×







1154












1155



        for _, policy := range policies {




×







1156



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




×







1157






×







1158



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




×







1159



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




×







1160



                        return err




×







1161



                }




×







1162



        }










1163



        return nil




×







1164



}
























    

  • 

    •