kubeovn / kube-ovn / 16336988527

coverage: 21.546%. Remained the same
16336988527

push

github

oilbeater 
fix: only addOrUpdateSubnetQueue if the GatewayType is distributed instead of if it's not centralized (#5493)

Signed-off-by: Joachim Hill-Grannec <joachim.hill@gmail.com>
(cherry picked from commit a29592574)

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

2 existing lines in 1 file now uncovered.

10506 of 48760 relevant lines covered (21.55%)

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




×







74



        node := obj.(*v1.Node)




×







75



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




×







76



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




×







77



        c.deletingNodeObjMap.Store(key, node)




×







78



        c.deleteNodeQueue.Add(key)




×







79



}




×







80












81



func nodeUnderlayAddressSetName(node string, af int) string {




×







82



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




×







83



}




×







84












85



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




×







86



        c.nodeKeyMutex.LockKey(key)




×







87



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




×







88












89



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




×







90



        if err != nil {




×







91



                if k8serrors.IsNotFound(err) {




×







92



                        return nil




×







93



                }




×







94



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




×







95



                return err




×







96



        }










97



        node := cachedNode.DeepCopy()




×







98



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




×







99






×







100



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




×







101



        if err != nil {




×







102



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




×







103



                return err




×







104



        }




×







105












106



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




×







107



        for _, subnet := range subnets {




×







108



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




×







109



                        continue




×







110



                }










111












112



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




×







113



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




×







114



                        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)




×







115



                        klog.Warning(msg)




×







116



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




×







117



                        break




×







118



                }










119



        }










120












121



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




×







122



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




×







123



                return err




×







124



        }




×







125












126



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




×







127



        if err != nil {




×







128



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




×







129



                return err




×







130



        }




×







131












132



        var v4IP, v6IP, mac string




×







133






×







134



        portName := util.NodeLspName(key)




×







135



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




×







136



                macStr := node.Annotations[util.MacAddressAnnotation]




×







137



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




×







138



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




×







139



                if err != nil {




×







140



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




×







141



                        return err




×







142



                }




×







143



        } else {




×







144



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




×







145



                if err != nil {




×







146



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




×







147



                        return err




×







148



                }




×







149












150



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










151



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




×







152



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




×







153



                        return err




×







154



                }




×







155



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




×







156



        }










157












158



        ipStr := util.GetStringIP(v4IP, v6IP)




×







159



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




×







160



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




×







161



                return err




×







162



        }




×







163












164



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




×







165



                if ip == "" {




×







166



                        continue




×







167



                }










168












169



                nodeIP, af := nodeIPv4, 4




×







170



                protocol := util.CheckProtocol(ip)




×







171



                if protocol == kubeovnv1.ProtocolIPv6 {




×







172



                        nodeIP, af = nodeIPv6, 6




×







173



                }




×







174



                if nodeIP != "" {




×







175



                        var (




×







176



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




×







177



                                action      = kubeovnv1.PolicyRouteActionReroute




×







178



                                externalIDs = map[string]string{




×







179



                                        "vendor":         util.CniTypeName,




×







180



                                        "node":           node.Name,




×







181



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




×







182



                                }




×







183



                        )




×







184



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




×







185



                        if err = c.addPolicyRouteToVpc(




×







186



                                c.config.ClusterRouter,




×







187



                                &kubeovnv1.PolicyRoute{




×







188



                                        Priority:  util.NodeRouterPolicyPriority,




×







189



                                        Match:     match,




×







190



                                        Action:    action,




×







191



                                        NextHopIP: ip,




×







192



                                },




×







193



                                externalIDs,




×







194



                        ); err != nil {




×







195



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




×







196



                                return err




×







197



                        }




×







198












199



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




×







200



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




×







201



                                if util.CheckProtocol(ip) == protocol {




×







202



                                        dnsIPs = append(dnsIPs, ip)




×







203



                                }




×







204



                        }










205












206



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




×







207



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




×







208



                                return err




×







209



                        }




×







210



                }










211



        }










212












213



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




×







214



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




×







215



                return err




×







216



        }




×







217












218



        patch := util.KVPatch{




×







219



                util.IPAddressAnnotation:     ipStr,




×







220



                util.MacAddressAnnotation:    mac,




×







221



                util.CidrAnnotation:          subnet.Spec.CIDRBlock,




×







222



                util.GatewayAnnotation:       subnet.Spec.Gateway,




×







223



                util.LogicalSwitchAnnotation: c.config.NodeSwitch,




×







224



                util.AllocatedAnnotation:     "true",




×







225



                util.PortNameAnnotation:      portName,




×







226



        }




×







227



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




×







228



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




×







229



                return err




×







230



        }




×







231












232



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




×







233



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




×







234



                return err




×







235



        }




×







236












237



        for _, subnet := range subnets {




×







238



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




×







239



                        continue




×







240



                }










241



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




×







242



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




×







243



                        return err




×







244



                }




×







245



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










246



                c.addOrUpdateSubnetQueue.Add(subnet.Name)




×







247



        }










248












249



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










250



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




×







251



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




×







252



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




×







253



                return err




×







254



        }




×







255












256



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




×







257



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




×







258



                return err




×







259



        }




×







260












261



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




×







262



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




×







263



                return err




×







264



        }




×







265












266



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




×







267



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




×







268



                return err




×







269



        }




×







270



        return nil




×







271



}










272












273



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




×







274



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




×







275



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




×







276



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




×







277



                return err




×







278



        }




×







279












280



        for _, pn := range providerNetworks {




×







281



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




×







282



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




×







283






×







284



                var newPn *kubeovnv1.ProviderNetwork




×







285



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




×







286



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




×







287



                        newPn = pn.DeepCopy()




×







288



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




×







289



                        excluded = true




×







290



                }




×







291












292



                var customInterface string




×







293



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




×







294



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




×







295



                                customInterface = v.Interface




×







296



                                break




×







297



                        }










298



                }










299



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




×







300



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




×







301



                                if newPn == nil {




×







302



                                        newPn = pn.DeepCopy()




×







303



                                }




×







304



                                var index *int




×







305



                                for i := range newPn.Spec.CustomInterfaces {




×







306



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




×







307



                                                index = &i




×







308



                                                break




×







309



                                        }










310



                                }










311



                                if index != nil {




×







312



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




×







313



                                } else {




×







314



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




×







315



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




×







316



                                }




×







317



                        }










318



                }










319












320



                if newPn != nil {




×







321



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




×







322



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




×







323



                                return err




×







324



                        }




×







325



                }










326












327



                if len(node.Annotations) != 0 {




×







328



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




×







329



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




×







330



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




×







331



                                return err




×







332



                        }




×







333



                }










334












335



                if excluded {




×







336



                        if newPn == nil {




×







337



                                newPn = pn.DeepCopy()




×







338



                        } else {




×







339



                                newPn = newPn.DeepCopy()




×







340



                        }




×







341












342



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




×







343



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




×







344



                                if err != nil {




×







345



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




×







346



                                        return err




×







347



                                }




×







348



                        }










349



                }










350



        }










351












352



        return nil




×







353



}










354












355



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




×







356



        c.nodeKeyMutex.LockKey(key)




×







357



        defer func() {




×







358



                _ = c.nodeKeyMutex.UnlockKey(key)




×







359



                if err == nil {




×







360



                        c.deletingNodeObjMap.Delete(key)




×







361



                }




×







362



        }()










363



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




×







364






×







365



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




×







366



        if !ok {




×







367



                return nil




×







368



        }




×







369



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




×







370



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




×







371



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




×







372



                return nil




×







373



        }




×







374



        return c.deleteNode(key)




×







375



}










376












377



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




×







378



        portName := util.NodeLspName(key)




×







379



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




×







380



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




×







381



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




×







382



                return err




×







383



        }




×







384



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




×







385



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




×







386



                return err




×







387



        }




×







388












389



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




×







390



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




×







391



                        klog.Error(err)




×







392



                        return err




×







393



                }




×







394



        }










395












396



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










397



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




×







398



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




×







399



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




×







400



                return err




×







401



        }




×







402












403



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




×







404



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




×







405



                return err




×







406



        }




×







407












408



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




×







409



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




×







410



                return err




×







411



        }




×







412



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




×







413



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




×







414



                return err




×







415



        }




×







416












417



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




×







418



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




×







419






×







420



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




×







421



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




×







422



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




×







423



                return err




×







424



        }




×







425












426



        for _, pn := range providerNetworks {




×







427



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




×







428



                        klog.Error(err)




×







429



                        return err




×







430



                }




×







431



        }










432



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




×







433



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




×







434



                return err




×







435



        }




×







436












437



        return nil




×







438



}










439












440



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




×







441



        // update provider network status




×







442



        var needUpdate bool




×







443



        newPn := pn.DeepCopy()




×







444



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




×







445



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




×







446



                needUpdate = true




×







447



        }




×







448



        if newPn.Status.RemoveNodeConditions(node) {




×







449



                needUpdate = true




×







450



        }




×







451



        if needUpdate {




×







452



                var err error




×







453



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




×







454



                if err != nil {




×







455



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




×







456



                        return err




×







457



                }




×







458



        }










459












460



        // update provider network spec










461



        pn, newPn = newPn, nil




×







462



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




×







463



                newPn = pn.DeepCopy()




×







464



                newPn.Spec.ExcludeNodes = excludeNodes




×







465



        }




×







466












467



        var changed bool




×







468



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




×







469



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




×







470



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




×







471



                if !changed {




×







472



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




×







473



                }




×







474



                if len(nodes) != 0 {




×







475



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




×







476



                }




×







477



        }










478



        if changed {




×







479



                newPn = pn.DeepCopy()




×







480



                newPn.Spec.CustomInterfaces = customInterfaces




×







481



        }




×







482



        if newPn != nil {




×







483



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




×







484



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




×







485



                        return err




×







486



                }




×







487



        }










488












489



        return nil




×







490



}










491












492



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




×







493



        c.nodeKeyMutex.LockKey(key)




×







494



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




×







495



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




×







496






×







497



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




×







498



        if err != nil {




×







499



                if k8serrors.IsNotFound(err) {




×







500



                        return nil




×







501



                }




×







502



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




×







503



                return err




×







504



        }










505












506



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




×







507



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




×







508



                return err




×







509



        }




×







510












511



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




×







512



        if err != nil {




×







513



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




×







514



                return err




×







515



        }




×







516












517



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




×







518



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




×







519



                return err




×







520



        }




×







521



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




×







522



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




×







523



                return err




×







524



        }




×







525












526



        for _, cachedSubnet := range subnets {




×








NEW


527



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




×







528



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




×







529



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




×







530



                        c.addOrUpdateSubnetQueue.Add(cachedSubnet.Name)




×







531



                        continue




×







532



                }










533



                subnet := cachedSubnet.DeepCopy()




×







534



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




×







535



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




×







536



                                klog.Error(err)




×







537



                                return err




×







538



                        }




×







539



                }










540



        }










541












542



        return nil




×







543



}










544












545



func (c *Controller) checkSubnetGateway() {




×







546



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




×







547



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




×







548



        }




×







549



}










550












551



func (c *Controller) checkSubnetGatewayNode() error {




×







552



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




×







553



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




×







554



        if err != nil {




×







555



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




×







556



                return err




×







557



        }




×







558



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




×







559



        if err != nil {




×







560



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




×







561



                return err




×







562



        }




×







563












564



        for _, subnet := range subnetList {




×







565



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




×







566



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




×







567



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




×







568



                        subnet.Spec.GatewayNode == "" ||




×







569



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




×







570



                        !subnet.Spec.EnableEcmp {




×







571



                        continue




×







572



                }










573



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




×







574



                if len(gwNodes) < 2 {




×







575



                        continue




×







576



                }










577












578



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




×







579



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




×







580



                        if err != nil {




×







581



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




×







582



                                continue




×







583



                        }










584



                        for _, node := range nodes {




×







585



                                ipStr := node.Annotations[util.IPAddressAnnotation]




×







586



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




×







587



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




×







588



                                                continue




×







589



                                        }










590












591



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




×







592



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




×







593



                                                pinger, err := goping.NewPinger(ip)




×







594



                                                if err != nil {




×







595



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




×







596



                                                }




×







597



                                                pinger.SetPrivileged(true)




×







598






×







599



                                                count := 5




×







600



                                                pinger.Count = count




×







601



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




×







602



                                                pinger.Interval = 1 * time.Second




×







603






×







604



                                                var pingSucceeded bool




×







605



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




×







606



                                                        pingSucceeded = true




×







607



                                                        pinger.Stop()




×







608



                                                }




×







609



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




×







610



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




×







611



                                                        return err




×







612



                                                }




×







613












614



                                                nodeIsReady := nodeReady(node)




×







615



                                                if !pingSucceeded || !nodeIsReady {




×







616



                                                        if exist {




×







617



                                                                if !pingSucceeded {




×







618



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




×







619



                                                                }




×







620



                                                                if !nodeIsReady {




×







621



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




×







622



                                                                }




×







623



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




×







624



                                                                nextHops.Remove(ip)




×







625



                                                                delete(nameIPMap, node.Name)




×







626



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




×







627



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




×







628



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




×







629



                                                                        return err




×







630



                                                                }




×







631



                                                        }










632



                                                } else {




×







633



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




×







634



                                                        if !exist {




×







635



                                                                nextHops.Add(ip)




×







636



                                                                if nameIPMap == nil {




×







637



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




×







638



                                                                }




×







639



                                                                nameIPMap[node.Name] = ip




×







640



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




×







641



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




×







642



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




×







643



                                                                        return err




×







644



                                                                }




×







645



                                                        }










646



                                                }










647



                                        } else if exist {




×







648



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




×







649



                                                nextHops.Remove(ip)




×







650



                                                delete(nameIPMap, node.Name)




×







651



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




×







652



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




×







653



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




×







654



                                                        return err




×







655



                                                }




×







656



                                        }










657



                                }










658



                        }










659



                }










660



        }










661



        return nil




×







662



}










663












664



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




×







665



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




×







666



        var err error




×







667



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




×







668



                return nil




×







669



        }




×







670



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




×







671



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




×







672



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




×







673



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




×







674



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




×







675



                        return err




×







676



                }




×







677



        }










678



        return err




×







679



}










680












681



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




×







682



        i := 0




×







683



        for ; ; i++ {




×







684



                err = f(node)




×







685



                if err == nil {




×







686



                        return




×







687



                }




×







688



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




×







689



                if i >= (attempts - 1) {




×







690



                        break




×







691



                }










692



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




×







693



        }










694



        if i >= (attempts - 1) {




×







695



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




×







696



                klog.Error(errMsg)




×







697



                return errMsg




×







698



        }




×







699



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




×







700



        return nil




×







701



}










702












703



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




×







704



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




×







705



        for _, pod := range pods {




×







706



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




×







707



                        continue




×







708



                }










709












710



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




×







711



                        subnetName := pod.Annotations[util.LogicalSwitchAnnotation]




×







712



                        if subnetName == "" {




×







713



                                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)




×







714



                                continue




×







715



                        }










716












717



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




×







718



                        if err != nil {




×







719



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




×







720



                                return nil, err




×







721



                        }




×







722












723



                        if subnet.Spec.Vlan == "" {




×







724



                                continue




×







725



                        }










726



                }










727












728



                podName := c.getNameByPod(pod)




×







729






×







730



                podNets, err := c.getPodKubeovnNets(pod)




×







731



                if err != nil {




×







732



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




×







733



                        return nil, err




×







734



                }




×







735












736



                for _, podNet := range podNets {




×







737



                        if !isOvnSubnet(podNet.Subnet) {




×







738



                                continue




×







739



                        }










740












741



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




×







742



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




×







743



                        }




×







744



                }










745



        }










746



        return ports, nil




×







747



}










748












749



func (c *Controller) CheckNodePortGroup() {




×







750



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




×







751



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




×







752



        }




×







753



}










754












755



func (c *Controller) checkAndUpdateNodePortGroup() error {




×







756



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




×







757



        var networkPolicyExists bool




×







758



        if c.config.EnableNP {




×







759



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




×







760



                networkPolicyExists = len(np) != 0




×







761



        }




×







762












763



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




×







764



        if err != nil {




×







765



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




×







766



                return err




×







767



        }




×







768












769



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




×







770



        if err != nil {




×







771



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




×







772



                return err




×







773



        }




×







774












775



        for _, node := range nodes {




×







776



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




×







777



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




×







778






×







779



                // use join IP only when no internal IP exists




×







780



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




×







781



                joinIP := node.Annotations[util.IPAddressAnnotation]




×







782



                joinIPv4, joinIPv6 := util.SplitStringIP(joinIP)




×







783



                if nodeIPv4 == "" {




×







784



                        nodeIPv4 = joinIPv4




×







785



                }




×







786



                if nodeIPv6 == "" {




×







787



                        nodeIPv6 = joinIPv6




×







788



                }




×







789



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




×







790






×







791



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




×







792



                if err != nil {




×







793



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




×







794



                        return err




×







795



                }




×







796












797



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




×







798



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




×







799



                        return err




×







800



                }




×







801












802



                if networkPolicyExists {




×







803



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




×







804



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




×







805



                        }




×







806



                } else {




×







807



                        // clear all acl




×







808



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




×







809



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




×







810



                        }




×







811



                }










812



        }










813












814



        return nil




×







815



}










816












817



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




×







818



        annoChassisName := node.Annotations[util.ChassisAnnotation]




×







819



        if annoChassisName == "" {




×







820



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




×







821



                return nil




×







822



        }




×







823



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




×







824



        if err != nil {




×







825



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




×







826



                return err




×







827



        }




×







828



        if chassis == nil {




×







829



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




×







830



                // chassis name conflict, do GC




×







831



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




×







832



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




×







833



                        return err




×







834



                }




×







835



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




×







836



                klog.Error(err)




×







837



                return err




×







838



        }










839












840



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




×







841



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




×







842



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




×







843



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




×







844



                        klog.Error(err)




×







845



                        return err




×







846



                }




×







847



        }










848



        return nil




×







849



}










850












851



func (c *Controller) addNodeGatewayStaticRoute() error {




×







852



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




×







853



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




×







854



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




×







855



                if err != nil {




×







856



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




×







857



                }




×







858



                if len(existRoute) != 0 {




×







859



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




×







860



                        return nil




×







861



                }




×







862



        }










863



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




×







864



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




×







865



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




×







866



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




×







867



                                continue




×







868



                        }










869












870



                        if err := c.addStaticRouteToVpc(




×







871



                                c.config.ClusterRouter,




×







872



                                &kubeovnv1.StaticRoute{




×







873



                                        Policy:     kubeovnv1.PolicyDst,




×







874



                                        CIDR:       cidrBlock,




×







875



                                        NextHopIP:  nextHop,




×







876



                                        RouteTable: util.MainRouteTable,




×







877



                                },




×







878



                        ); err != nil {




×







879



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




×







880



                                return err




×







881



                        }




×







882



                }










883



        }










884



        return nil




×







885



}










886












887



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




×







888



        ipSuffix := "ip4"




×







889



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




×







890



                ipSuffix = "ip6"




×







891



        }




×







892



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




×







893



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




×







894



        if err != nil {




×







895



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




×







896



                return nil, nil, err




×







897



        }




×







898



        if len(policyList) == 0 {




×







899



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




×







900



        }




×







901



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




×







902



}










903












904



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




×







905



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




×







906



        if err != nil {




×







907



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




×







908



                return err




×







909



        }




×







910












911



        addresses := c.ipam.GetPodAddress(portName)




×







912



        for _, addr := range addresses {




×







913



                if addr.IP == "" {




×







914



                        continue




×







915



                }










916



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




×







917



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




×







918



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




×







919



                        return err




×







920



                }




×







921



        }










922












923



        for _, subnet := range subnets {




×







924



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




×







925



                        continue




×







926



                }










927












928



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




×







929



                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, nodeName)




×







930



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




×







931



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




×







932



                                return err




×







933



                        }




×







934












935



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




×







936



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




×







937



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




×







938



                                return err




×







939



                        }




×







940



                }










941












942



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




×







943



                        if subnet.Spec.EnableEcmp {




×







944



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




×







945



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




×







946



                                        if err != nil {




×







947



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




×







948



                                                continue




×







949



                                        }










950












951



                                        exist := false




×







952



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




×







953



                                                exist = true




×







954



                                        }




×







955












956



                                        if exist {




×







957



                                                nextHops.Remove(nameIPMap[nodeName])




×







958



                                                delete(nameIPMap, nodeName)




×







959






×







960



                                                if nextHops.Size() == 0 {




×







961



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




×







962



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




×







963



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




×







964



                                                                return err




×







965



                                                        }




×







966



                                                } else {




×







967



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




×







968



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




×







969



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




×







970



                                                                return err




×







971



                                                        }




×







972



                                                }










973



                                        }










974



                                }










975



                        } else {




×







976



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




×







977



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




×







978



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




×







979



                                        return err




×







980



                                }




×







981



                        }










982



                }










983



        }










984



        return nil




×







985



}










986












987



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




×







988



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




×







989



        if err != nil {




×







990



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




×







991



                return err




×







992



        }




×







993












994



        for _, subnet := range subnets {




×







995



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




×







996



                        continue




×







997



                }










998












999



                if subnet.Spec.EnableEcmp {




×







1000



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




×







1001



                                continue




×







1002



                        }










1003












1004



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




×







1005



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




×







1006



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




×







1007



                                                continue




×







1008



                                        }










1009












1010



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




×







1011



                                        if err != nil {




×







1012



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




×







1013



                                                continue




×







1014



                                        }










1015



                                        if nameIPMap[nodeName] == nextHop {




×







1016



                                                continue




×







1017



                                        }










1018












1019



                                        nextHops.Add(nextHop)




×







1020



                                        if nameIPMap == nil {




×







1021



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




×







1022



                                        }




×







1023



                                        nameIPMap[nodeName] = nextHop




×







1024



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




×







1025



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




×







1026



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




×







1027



                                                return err




×







1028



                                        }




×







1029



                                }










1030



                        }










1031



                } else {




×







1032



                        if subnet.Status.ActivateGateway != nodeName {




×







1033



                                continue




×







1034



                        }










1035



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




×







1036



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




×







1037



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




×







1038



                                return err




×







1039



                        }




×







1040



                }










1041



        }










1042



        return nil




×







1043



}










1044












1045



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




×







1046



        if len(dnsIPs) == 0 {




×







1047



                return c.deletePolicyRouteForLocalDNSCacheOnNode(nodeName, af)




×







1048



        }




×







1049












1050



        var (




×







1051



                externalIDs = map[string]string{




×







1052



                        "vendor":          util.CniTypeName,




×







1053



                        "node":            nodeName,




×







1054



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




×







1055



                        "isLocalDnsCache": "true",




×







1056



                }




×







1057



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




×







1058



                action   = kubeovnv1.PolicyRouteActionReroute




×







1059



                nextHops = []string{nodeIP}




×







1060



        )




×







1061



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




×







1062



        for _, ip := range dnsIPs {




×







1063



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




×







1064



        }




×







1065












1066



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




×







1067



        if err != nil {




×







1068



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




×







1069



                return err




×







1070



        }




×







1071












1072



        for _, policy := range policies {




×







1073



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




×







1074



                        continue




×







1075



                }










1076



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




×







1077



                        matches.Remove(policy.Match)




×







1078



                        continue




×







1079



                }










1080



                // delete unused policy router policy










1081



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




×







1082



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




×







1083



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




×







1084



                        return err




×







1085



                }




×







1086



        }










1087












1088



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




×







1089



                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)




×







1090



                if err := c.addPolicyRouteToVpc(




×







1091



                        c.config.ClusterRouter,




×







1092



                        &kubeovnv1.PolicyRoute{




×







1093



                                Priority:  util.NodeRouterPolicyPriority,




×







1094



                                Match:     match,




×







1095



                                Action:    action,




×







1096



                                NextHopIP: nodeIP,




×







1097



                        },




×







1098



                        externalIDs,




×







1099



                ); err != nil {




×







1100



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




×







1101



                        return err




×







1102



                }




×







1103



        }










1104












1105



        return nil




×







1106



}










1107












1108



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




×







1109



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




×







1110



                "vendor":          util.CniTypeName,




×







1111



                "node":            nodeName,




×







1112



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




×







1113



                "isLocalDnsCache": "true",




×







1114



        }, true)




×







1115



        if err != nil {




×







1116



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




×







1117



                return err




×







1118



        }




×







1119



        if len(policies) == 0 {




×







1120



                return nil




×







1121



        }




×







1122












1123



        for _, policy := range policies {




×







1124



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




×







1125






×







1126



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




×







1127



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




×







1128



                        return err




×







1129



                }




×







1130



        }










1131



        return nil




×







1132



}
























    

  • 

    •