kubeovn / kube-ovn / 14569165934

coverage: 21.723% (-0.007%) from 21.73%
14569165934

push

github

web-flow 
controller: ensure ovn route policy is reconciled after node is initialized (#5166)

Signed-off-by: zhangzujian <zhangzujian.7@gmail.com>

0 of 6 new or added lines in 2 files covered. (0.0%)

3 existing lines in 2 files now uncovered.

10249 of 47181 relevant lines covered (21.72%)

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



        portName := util.NodeLspName(key)




×







134



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




×







135



                macStr := node.Annotations[util.MacAddressAnnotation]




×







136



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




×







137



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




×







138



                if err != nil {




×







139



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




×







140



                        return err




×







141



                }




×







142



        } else {




×







143



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




×







144



                if err != nil {




×







145



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




×







146



                        return err




×







147



                }




×







148



        }










149












150



        ipStr := util.GetStringIP(v4IP, v6IP)




×







151



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




×







152



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




×







153



                return err




×







154



        }




×







155












156



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




×







157



                if ip == "" {




×







158



                        continue




×







159



                }










160












161



                nodeIP, af := nodeIPv4, 4




×







162



                protocol := util.CheckProtocol(ip)




×







163



                if protocol == kubeovnv1.ProtocolIPv6 {




×







164



                        nodeIP, af = nodeIPv6, 6




×







165



                }




×







166



                if nodeIP != "" {




×







167



                        var (




×







168



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




×







169



                                action      = kubeovnv1.PolicyRouteActionReroute




×







170



                                externalIDs = map[string]string{




×







171



                                        "vendor":         util.CniTypeName,




×







172



                                        "node":           node.Name,




×







173



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




×







174



                                }




×







175



                        )




×







176



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




×







177



                        if err = c.addPolicyRouteToVpc(




×







178



                                c.config.ClusterRouter,




×







179



                                &kubeovnv1.PolicyRoute{




×







180



                                        Priority:  util.NodeRouterPolicyPriority,




×







181



                                        Match:     match,




×







182



                                        Action:    action,




×







183



                                        NextHopIP: ip,




×







184



                                },




×







185



                                externalIDs,




×







186



                        ); err != nil {




×







187



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




×







188



                                return err




×







189



                        }




×







190












191



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




×







192



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




×







193



                                if util.CheckProtocol(ip) == protocol {




×







194



                                        dnsIPs = append(dnsIPs, ip)




×







195



                                }




×







196



                        }










197












198



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




×







199



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




×







200



                                return err




×







201



                        }




×







202



                }










203



        }










204












205



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




×







206



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




×







207



                return err




×







208



        }




×







209












210



        patch := util.KVPatch{




×







211



                util.IPAddressAnnotation:     ipStr,




×







212



                util.MacAddressAnnotation:    mac,




×







213



                util.CidrAnnotation:          subnet.Spec.CIDRBlock,




×







214



                util.GatewayAnnotation:       subnet.Spec.Gateway,




×







215



                util.LogicalSwitchAnnotation: c.config.NodeSwitch,




×







216



                util.AllocatedAnnotation:     "true",




×







217



                util.PortNameAnnotation:      portName,




×







218



        }




×







219



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




×







220



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




×







221



                return err




×







222



        }




×







223












224



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




×







225



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




×







226



                return err




×







227



        }




×







228












229



        for _, subnet := range subnets {




×







230



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




×







231



                        continue




×







232



                }










233



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




×







234



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




×







235



                        return err




×







236



                }




×







237



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










238



                c.addOrUpdateSubnetQueue.Add(subnet.Name)




×







239



        }










240












241



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










242



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




×







243



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




×







244



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




×







245



                return err




×







246



        }




×







247












248



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




×







249



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




×







250



                return err




×







251



        }




×







252












253



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




×







254



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




×







255



                return err




×







256



        }




×







257












258



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




×







259



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




×







260



                return err




×







261



        }




×







262



        return nil




×







263



}










264












265



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




×







266



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




×







267



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




×







268



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




×







269



                return err




×







270



        }




×







271












272



        for _, pn := range providerNetworks {




×







273



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




×







274



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




×







275






×







276



                var newPn *kubeovnv1.ProviderNetwork




×







277



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




×







278



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




×







279



                        newPn = pn.DeepCopy()




×







280



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




×







281



                        excluded = true




×







282



                }




×







283












284



                var customInterface string




×







285



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




×







286



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




×







287



                                customInterface = v.Interface




×







288



                                break




×







289



                        }










290



                }










291



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




×







292



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




×







293



                                if newPn == nil {




×







294



                                        newPn = pn.DeepCopy()




×







295



                                }




×







296



                                var index *int




×







297



                                for i := range newPn.Spec.CustomInterfaces {




×







298



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




×







299



                                                index = &i




×







300



                                                break




×







301



                                        }










302



                                }










303



                                if index != nil {




×







304



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




×







305



                                } else {




×







306



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




×







307



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




×







308



                                }




×







309



                        }










310



                }










311












312



                if newPn != nil {




×







313



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




×







314



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




×







315



                                return err




×







316



                        }




×







317



                }










318












319



                if len(node.Annotations) != 0 {




×







320



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




×







321



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




×







322



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




×







323



                                return err




×







324



                        }




×







325



                }










326












327



                if excluded {




×







328



                        if newPn == nil {




×







329



                                newPn = pn.DeepCopy()




×







330



                        } else {




×







331



                                newPn = newPn.DeepCopy()




×







332



                        }




×







333












334



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




×







335



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




×







336



                                if err != nil {




×







337



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




×







338



                                        return err




×







339



                                }




×







340



                        }










341



                }










342



        }










343












344



        return nil




×







345



}










346












347



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




×







348



        c.nodeKeyMutex.LockKey(key)




×







349



        defer func() {




×







350



                _ = c.nodeKeyMutex.UnlockKey(key)




×







351



                if err == nil {




×







352



                        c.deletingNodeObjMap.Delete(key)




×







353



                }




×







354



        }()










355



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




×







356






×







357



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




×







358



        if !ok {




×







359



                return nil




×







360



        }




×







361



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




×







362



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




×







363



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




×







364



                return nil




×







365



        }




×







366



        return c.deleteNode(key)




×







367



}










368












369



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




×







370



        portName := util.NodeLspName(key)




×







371



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




×







372



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




×







373



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




×







374



                return err




×







375



        }




×







376



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




×







377



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




×







378



                return err




×







379



        }




×







380












381



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




×







382



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




×







383



                        klog.Error(err)




×







384



                        return err




×







385



                }




×







386



        }










387












388



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










389



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




×







390



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




×







391



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




×







392



                return err




×







393



        }




×







394












395



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




×







396



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




×







397



                return err




×







398



        }




×







399












400



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




×







401



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




×







402



                return err




×







403



        }




×







404



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




×







405



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




×







406



                return err




×







407



        }




×







408












409



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




×







410



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




×







411






×







412



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




×







413



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




×







414



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




×







415



                return err




×







416



        }




×







417












418



        for _, pn := range providerNetworks {




×







419



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




×







420



                        klog.Error(err)




×







421



                        return err




×







422



                }




×







423



        }










424



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




×







425



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




×







426



                return err




×







427



        }




×







428












429



        return nil




×







430



}










431












432



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




×







433



        // update provider network status




×







434



        var needUpdate bool




×







435



        newPn := pn.DeepCopy()




×







436



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




×







437



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




×







438



                needUpdate = true




×







439



        }




×







440



        if newPn.Status.RemoveNodeConditions(node) {




×







441



                needUpdate = true




×







442



        }




×







443



        if needUpdate {




×







444



                var err error




×







445



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




×







446



                if err != nil {




×







447



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




×







448



                        return err




×







449



                }




×







450



        }










451












452



        // update provider network spec










453



        pn, newPn = newPn, nil




×







454



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




×







455



                newPn = pn.DeepCopy()




×







456



                newPn.Spec.ExcludeNodes = excludeNodes




×







457



        }




×







458












459



        var changed bool




×







460



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




×







461



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




×







462



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




×







463



                if !changed {




×







464



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




×







465



                }




×







466



                if len(nodes) != 0 {




×







467



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




×







468



                }




×







469



        }










470



        if changed {




×







471



                newPn = pn.DeepCopy()




×







472



                newPn.Spec.CustomInterfaces = customInterfaces




×







473



        }




×







474



        if newPn != nil {




×







475



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




×







476



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




×







477



                        return err




×







478



                }




×







479



        }










480












481



        return nil




×







482



}










483












484



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




×







485



        c.nodeKeyMutex.LockKey(key)




×







486



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




×







487



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




×







488






×







489



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




×







490



        if err != nil {




×







491



                if k8serrors.IsNotFound(err) {




×







492



                        return nil




×







493



                }




×







494



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




×







495



                return err




×







496



        }










497












498



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




×







499



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




×







500



                return err




×







501



        }




×







502












503



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




×







504



        if err != nil {




×







505



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




×







506



                return err




×







507



        }




×







508












509



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




×







510



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




×







511



                return err




×







512



        }




×







513



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




×







514



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




×







515



                return err




×







516



        }




×







517












518



        for _, cachedSubnet := range subnets {




×








NEW


519



                if cachedSubnet.Spec.GatewayType != kubeovnv1.GWCentralizedType {




×








NEW


520



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




×








NEW


521



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




×








NEW


522



                        c.addOrUpdateSubnetQueue.Add(cachedSubnet.Name)




×








NEW


523



                        continue




×







524



                }










525



                subnet := cachedSubnet.DeepCopy()




×







526



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




×







527



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




×







528



                                klog.Error(err)




×







529



                                return err




×







530



                        }




×







531



                }










532



        }










533












534



        return nil




×







535



}










536












537



func (c *Controller) checkSubnetGateway() {




×







538



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




×







539



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




×







540



        }




×







541



}










542












543



func (c *Controller) checkSubnetGatewayNode() error {




×







544



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




×







545



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




×







546



        if err != nil {




×







547



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




×







548



                return err




×







549



        }




×







550



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




×







551



        if err != nil {




×







552



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




×







553



                return err




×







554



        }




×







555












556



        for _, subnet := range subnetList {




×







557



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




×







558



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




×







559



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




×







560



                        subnet.Spec.GatewayNode == "" ||




×







561



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




×







562



                        !subnet.Spec.EnableEcmp {




×







563



                        continue




×







564



                }










565



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




×







566



                if len(gwNodes) < 2 {




×







567



                        continue




×







568



                }










569












570



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




×







571



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




×







572



                        if err != nil {




×







573



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




×







574



                                continue




×







575



                        }










576



                        for _, node := range nodes {




×







577



                                ipStr := node.Annotations[util.IPAddressAnnotation]




×







578



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




×







579



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




×







580



                                                continue




×







581



                                        }










582












583



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




×







584



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




×







585



                                                pinger, err := goping.NewPinger(ip)




×







586



                                                if err != nil {




×







587



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




×







588



                                                }




×







589



                                                pinger.SetPrivileged(true)




×







590






×







591



                                                count := 5




×







592



                                                pinger.Count = count




×







593



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




×







594



                                                pinger.Interval = 1 * time.Second




×







595






×







596



                                                var pingSucceeded bool




×







597



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




×







598



                                                        pingSucceeded = true




×







599



                                                        pinger.Stop()




×







600



                                                }




×







601



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




×







602



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




×







603



                                                        return err




×







604



                                                }




×







605












606



                                                nodeIsReady := nodeReady(node)




×







607



                                                if !pingSucceeded || !nodeIsReady {




×







608



                                                        if exist {




×







609



                                                                if !pingSucceeded {




×







610



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




×







611



                                                                }




×







612



                                                                if !nodeIsReady {




×







613



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




×







614



                                                                }




×







615



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




×







616



                                                                nextHops.Remove(ip)




×







617



                                                                delete(nameIPMap, node.Name)




×







618



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




×







619



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




×







620



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




×







621



                                                                        return err




×







622



                                                                }




×







623



                                                        }










624



                                                } else {




×







625



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




×







626



                                                        if !exist {




×







627



                                                                nextHops.Add(ip)




×







628



                                                                if nameIPMap == nil {




×







629



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




×







630



                                                                }




×







631



                                                                nameIPMap[node.Name] = ip




×







632



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




×







633



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




×







634



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




×







635



                                                                        return err




×







636



                                                                }




×







637



                                                        }










638



                                                }










639



                                        } else if exist {




×







640



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




×







641



                                                nextHops.Remove(ip)




×







642



                                                delete(nameIPMap, node.Name)




×







643



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




×







644



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




×







645



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




×







646



                                                        return err




×







647



                                                }




×







648



                                        }










649



                                }










650



                        }










651



                }










652



        }










653



        return nil




×







654



}










655












656



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




×







657



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




×







658



        var err error




×







659



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




×







660



                return nil




×







661



        }




×







662



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




×







663



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




×







664



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




×







665



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




×







666



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




×







667



                        return err




×







668



                }




×







669



        }










670



        return err




×







671



}










672












673



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




×







674



        i := 0




×







675



        for ; ; i++ {




×







676



                err = f(node)




×







677



                if err == nil {




×







678



                        return




×







679



                }




×







680



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




×







681



                if i >= (attempts - 1) {




×







682



                        break




×







683



                }










684



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




×







685



        }










686



        if i >= (attempts - 1) {




×







687



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




×







688



                klog.Error(errMsg)




×







689



                return errMsg




×







690



        }




×







691



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




×







692



        return nil




×







693



}










694












695



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




×







696



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




×







697



        for _, pod := range pods {




×







698



                if !isPodAlive(pod) || pod.Spec.HostNetwork || pod.Spec.NodeName != nodeName || pod.Annotations[util.LogicalRouterAnnotation] != c.config.ClusterRouter {




×







699



                        continue




×







700



                }










701



                podName := c.getNameByPod(pod)




×







702






×







703



                podNets, err := c.getPodKubeovnNets(pod)




×







704



                if err != nil {




×







705



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




×







706



                        return nil, err




×







707



                }




×







708












709



                for _, podNet := range podNets {




×







710



                        if !isOvnSubnet(podNet.Subnet) {




×







711



                                continue




×







712



                        }










713












714



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




×







715



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




×







716



                        }




×







717



                }










718



        }










719



        return ports, nil




×







720



}










721












722



func (c *Controller) CheckNodePortGroup() {




×







723



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




×







724



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




×







725



        }




×







726



}










727












728



func (c *Controller) checkAndUpdateNodePortGroup() error {




×







729



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




×







730



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




×







731



        networkPolicyExists := len(np) != 0




×







732






×







733



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




×







734



        if err != nil {




×







735



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




×







736



                return err




×







737



        }




×







738












739



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




×







740



        if err != nil {




×







741



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




×







742



                return err




×







743



        }




×







744












745



        for _, node := range nodes {




×







746



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




×







747



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




×







748






×







749



                // use join IP only when no internal IP exists




×







750



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




×







751



                joinIP := node.Annotations[util.IPAddressAnnotation]




×







752



                joinIPv4, joinIPv6 := util.SplitStringIP(joinIP)




×







753



                if nodeIPv4 == "" {




×







754



                        nodeIPv4 = joinIPv4




×







755



                }




×







756



                if nodeIPv6 == "" {




×







757



                        nodeIPv6 = joinIPv6




×







758



                }




×







759



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




×







760






×







761



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




×







762



                if err != nil {




×







763



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




×







764



                        return err




×







765



                }




×







766












767



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




×







768



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




×







769



                        return err




×







770



                }




×







771












772



                if networkPolicyExists {




×







773



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




×







774



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




×







775



                        }




×







776



                } else {




×







777



                        // clear all acl




×







778



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




×







779



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




×







780



                        }




×







781



                }










782



        }










783












784



        return nil




×







785



}










786












787



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




×







788



        annoChassisName := node.Annotations[util.ChassisAnnotation]




×







789



        if annoChassisName == "" {




×







790



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




×







791



                return nil




×







792



        }




×







793



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




×







794



        if err != nil {




×







795



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




×







796



                return err




×







797



        }




×







798



        if chassis == nil {




×







799



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




×







800



                // chassis name conflict, do GC




×







801



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




×







802



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




×







803



                        return err




×







804



                }




×







805



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




×







806



                klog.Error(err)




×







807



                return err




×







808



        }










809












810



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




×







811



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




×







812



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




×







813



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




×







814



                        klog.Error(err)




×







815



                        return err




×







816



                }




×







817



        }










818



        return nil




×







819



}










820












821



func (c *Controller) addNodeGatewayStaticRoute() error {




×







822



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




×







823



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




×







824



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




×







825



                if err != nil {




×







826



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




×







827



                }




×







828



                if len(existRoute) != 0 {




×







829



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




×







830



                        return nil




×







831



                }




×







832



        }










833



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




×







834



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




×







835



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




×







836



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




×







837



                                continue




×







838



                        }










839












840



                        if err := c.addStaticRouteToVpc(




×







841



                                c.config.ClusterRouter,




×







842



                                &kubeovnv1.StaticRoute{




×







843



                                        Policy:     kubeovnv1.PolicyDst,




×







844



                                        CIDR:       cidrBlock,




×







845



                                        NextHopIP:  nextHop,




×







846



                                        RouteTable: util.MainRouteTable,




×







847



                                },




×







848



                        ); err != nil {




×







849



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




×







850



                                return err




×







851



                        }




×







852



                }










853



        }










854



        return nil




×







855



}










856












857



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




×







858



        ipSuffix := "ip4"




×







859



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




×







860



                ipSuffix = "ip6"




×







861



        }




×







862



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




×







863



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




×







864



        if err != nil {




×







865



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




×







866



                return nil, nil, err




×







867



        }




×







868



        if len(policyList) == 0 {




×







869



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




×







870



        }




×







871



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




×







872



}










873












874



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




×







875



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




×







876



        if err != nil {




×







877



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




×







878



                return err




×







879



        }




×







880












881



        addresses := c.ipam.GetPodAddress(portName)




×







882



        for _, addr := range addresses {




×







883



                if addr.IP == "" {




×







884



                        continue




×







885



                }










886



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




×







887



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




×







888



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




×







889



                        return err




×







890



                }




×







891



        }










892












893



        for _, subnet := range subnets {




×







894



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




×







895



                        continue




×







896



                }










897












898



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




×







899



                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, nodeName)




×







900



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




×







901



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




×







902



                                return err




×







903



                        }




×







904












905



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




×







906



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




×







907



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




×







908



                                return err




×







909



                        }




×







910



                }










911












912



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




×







913



                        if subnet.Spec.EnableEcmp {




×







914



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




×







915



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




×







916



                                        if err != nil {




×







917



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




×







918



                                                continue




×







919



                                        }










920












921



                                        exist := false




×







922



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




×







923



                                                exist = true




×







924



                                        }




×







925












926



                                        if exist {




×







927



                                                nextHops.Remove(nameIPMap[nodeName])




×







928



                                                delete(nameIPMap, nodeName)




×







929






×







930



                                                if nextHops.Size() == 0 {




×







931



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




×







932



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




×







933



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




×







934



                                                                return err




×







935



                                                        }




×







936



                                                } else {




×







937



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




×







938



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




×







939



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




×







940



                                                                return err




×







941



                                                        }




×







942



                                                }










943



                                        }










944



                                }










945



                        } else {




×







946



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




×







947



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




×







948



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




×







949



                                        return err




×







950



                                }




×







951



                        }










952



                }










953



        }










954



        return nil




×







955



}










956












957



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




×







958



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




×







959



        if err != nil {




×







960



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




×







961



                return err




×







962



        }




×







963












964



        for _, subnet := range subnets {




×







965



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




×







966



                        continue




×







967



                }










968












969



                if subnet.Spec.EnableEcmp {




×







970



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




×







971



                                continue




×







972



                        }










973












974



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




×







975



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




×







976



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




×







977



                                                continue




×







978



                                        }










979












980



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




×







981



                                        if err != nil {




×







982



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




×







983



                                                continue




×







984



                                        }










985



                                        if nameIPMap[nodeName] == nextHop {




×







986



                                                continue




×







987



                                        }










988












989



                                        nextHops.Add(nextHop)




×







990



                                        if nameIPMap == nil {




×







991



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




×







992



                                        }




×







993



                                        nameIPMap[nodeName] = nextHop




×







994



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




×







995



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




×







996



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




×







997



                                                return err




×







998



                                        }




×







999



                                }










1000



                        }










1001



                } else {




×







1002



                        if subnet.Status.ActivateGateway != nodeName {




×







1003



                                continue




×







1004



                        }










1005



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




×







1006



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




×







1007



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




×







1008



                                return err




×







1009



                        }




×







1010



                }










1011



        }










1012



        return nil




×







1013



}










1014












1015



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




×







1016



        if len(dnsIPs) == 0 {




×







1017



                return c.deletePolicyRouteForLocalDNSCacheOnNode(nodeName, af)




×







1018



        }




×







1019












1020



        var (




×







1021



                externalIDs = map[string]string{




×







1022



                        "vendor":          util.CniTypeName,




×







1023



                        "node":            nodeName,




×







1024



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




×







1025



                        "isLocalDnsCache": "true",




×







1026



                }




×







1027



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




×







1028



                action   = kubeovnv1.PolicyRouteActionReroute




×







1029



                nextHops = []string{nodeIP}




×







1030



        )




×







1031



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




×







1032



        for _, ip := range dnsIPs {




×







1033



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




×







1034



        }




×







1035












1036



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




×







1037



        if err != nil {




×







1038



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




×







1039



                return err




×







1040



        }




×







1041












1042



        for _, policy := range policies {




×







1043



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




×







1044



                        continue




×







1045



                }










1046



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




×







1047



                        matches.Remove(policy.Match)




×







1048



                        continue




×







1049



                }










1050



                // delete unused policy router policy










1051



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




×







1052



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




×







1053



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




×







1054



                        return err




×







1055



                }




×







1056



        }










1057












1058



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




×







1059



                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)




×







1060



                if err := c.addPolicyRouteToVpc(




×







1061



                        c.config.ClusterRouter,




×







1062



                        &kubeovnv1.PolicyRoute{




×







1063



                                Priority:  util.NodeRouterPolicyPriority,




×







1064



                                Match:     match,




×







1065



                                Action:    action,




×







1066



                                NextHopIP: nodeIP,




×







1067



                        },




×







1068



                        externalIDs,




×







1069



                ); err != nil {




×







1070



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




×







1071



                        return err




×







1072



                }




×







1073



        }










1074












1075



        return nil




×







1076



}










1077












1078



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




×







1079



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




×







1080



                "vendor":          util.CniTypeName,




×







1081



                "node":            nodeName,




×







1082



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




×







1083



                "isLocalDnsCache": "true",




×







1084



        }, true)




×







1085



        if err != nil {




×







1086



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




×







1087



                return err




×







1088



        }




×







1089



        if len(policies) == 0 {




×







1090



                return nil




×







1091



        }




×







1092












1093



        for _, policy := range policies {




×







1094



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




×







1095






×







1096



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




×







1097



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




×







1098



                        return err




×







1099



                }




×







1100



        }










1101



        return nil




×







1102



}
























    

  • 

    •