Skip to content

Commit

Permalink
fix reconcile routes (kubeovn#4168)
Browse files Browse the repository at this point in the history 
Signed-off-by: zhangzujian <[email protected]>
  • Loading branch information
@zhangzujian
zhangzujian committed Jun 15, 2024
1 parent 25b6f47 commit 772c19c
Show file tree
Hide file tree
Showing 3 changed files with 91 additions and 41 deletions.
107 changes: 70 additions & 37 deletions pkg/daemon/controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
ovsutil "github.com/digitalocean/go-openvswitch/ovs"
"github.com/kubeovn/go-iptables/iptables"
"github.com/vishvananda/netlink"
"golang.org/x/sys/unix"
v1 "k8s.io/api/core/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand Down Expand Up @@ -681,27 +682,44 @@ func (c *Controller) reconcileRouters(event subnetEvent) error {
}
}

node, err := c.nodesLister.Get(c.config.NodeName)
if err != nil {
klog.Errorf("failed to get node %s %v", c.config.NodeName, err)
return err
}
nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
var joinIPv4, joinIPv6 string
if len(node.Annotations) != 0 {
joinIPv4, joinIPv6 = util.SplitStringIP(node.Annotations[util.IpAddressAnnotation])
}

joinCIDR := make([]string, 0, 2)
cidrs := make([]string, 0, len(subnets)*2)
for _, subnet := range subnets {
// The route for overlay subnet cidr via ovn0 should not be deleted even though subnet.Status has changed to not ready
if subnet.Spec.Vlan != "" || subnet.Spec.Vpc != c.config.ClusterRouter {
if subnet.Spec.Vlan != "" || subnet.Spec.Vpc != c.config.ClusterRouter ||
(subnet.Name != c.config.NodeSwitch && !subnet.Status.IsReady()) {
continue
}

for _, cidrBlock := range strings.Split(subnet.Spec.CIDRBlock, ",") {
if _, ipNet, err := net.ParseCIDR(cidrBlock); err != nil {
klog.Errorf("%s is not a valid cidr block", cidrBlock)
} else {
if nodeIPv4 != "" && util.CIDRContainIP(cidrBlock, nodeIPv4) {
continue
}
if nodeIPv6 != "" && util.CIDRContainIP(cidrBlock, nodeIPv6) {
continue
}
cidrs = append(cidrs, ipNet.String())
if subnet.Name == c.config.NodeSwitch {
joinCIDR = append(joinCIDR, ipNet.String())
}
}
}
}

node, err := c.nodesLister.Get(c.config.NodeName)
if err != nil {
klog.Errorf("failed to get node %s %v", c.config.NodeName, err)
return err
}
gateway, ok := node.Annotations[util.GatewayAnnotation]
if !ok {
klog.Errorf("annotation for node %s ovn.kubernetes.io/gateway not exists", node.Name)
Expand All @@ -713,42 +731,25 @@ func (c *Controller) reconcileRouters(event subnetEvent) error {
return fmt.Errorf("failed to get nic %s", util.NodeNic)
}

existRoutes, err := getNicExistRoutes(nic, gateway)
nodeNicRoutes, err := getNicExistRoutes(nic, gateway)
if err != nil {
klog.Error(err)
return err
}

toAdd, toDel := routeDiff(existRoutes, cidrs)
toAdd, toDel := routeDiff(nodeNicRoutes, cidrs, joinCIDR, joinIPv4, joinIPv6, gateway)
for _, r := range toDel {
_, cidr, _ := net.ParseCIDR(r)
if err = netlink.RouteDel(&netlink.Route{Dst: cidr}); err != nil {
if err = netlink.RouteDel(&netlink.Route{Dst: r.Dst}); err != nil {
klog.Errorf("failed to del route %v", err)
}
}

for _, r := range toAdd {
_, cidr, _ := net.ParseCIDR(r)
for _, gw := range strings.Split(gateway, ",") {
if util.CheckProtocol(gw) != util.CheckProtocol(r) {
continue
}
if err = netlink.RouteReplace(&netlink.Route{Dst: cidr, LinkIndex: nic.Attrs().Index, Scope: netlink.SCOPE_UNIVERSE, Gw: net.ParseIP(gw)}); err != nil {
klog.Errorf("failed to add route %v", err)
}
r.LinkIndex = nic.Attrs().Index
if err = netlink.RouteReplace(&r); err != nil {
klog.Errorf("failed to replace route %v: %v", r, err)
}
}

if oldSubnet != nil && newSubnet != nil && oldSubnet.Spec.HtbQos != "" && newSubnet.Spec.HtbQos == "" {
if err := c.deleteSubnetQos(newSubnet); err != nil {
klog.Errorf("failed to delete htb qos for subnet %s: %v", newSubnet.Name, err)
return err
}
} else if newSubnet != nil && newSubnet.Spec.HtbQos != "" {
if err := c.setSubnetQosPriority(newSubnet); err != nil {
klog.Errorf("failed to set htb qos priority for subnet %s: %v", newSubnet.Name, err)
return err
}
}
return nil
}

Expand All @@ -769,7 +770,10 @@ func getNicExistRoutes(nic netlink.Link, gateway string) ([]netlink.Route, error
return existRoutes, nil
}

func routeDiff(existRoutes []netlink.Route, cidrs []string) (toAdd, toDel []string) {
func routeDiff(existRoutes []netlink.Route, cidrs, joinCIDR []string, joinIPv4, joinIPv6, gateway string) (toAdd, toDel []netlink.Route) {
// joinIPv6 is not used for now
_ = joinIPv6

for _, route := range existRoutes {
if route.Scope == netlink.SCOPE_LINK || route.Dst == nil || route.Dst.IP.IsLinkLocalUnicast() {
continue
Expand All @@ -783,13 +787,15 @@ func routeDiff(existRoutes []netlink.Route, cidrs []string) (toAdd, toDel []stri
}
}
if !found {
toDel = append(toDel, route.Dst.String())
toDel = append(toDel, route)
}
}
if len(toDel) > 0 {
klog.Infof("route to del %v", toDel)
klog.Infof("routes to delete: %v", toDel)
}

ipv4, ipv6 := util.SplitStringIP(gateway)
gwV4, gwV6 := net.ParseIP(ipv4), net.ParseIP(ipv6)
for _, c := range cidrs {
found := false
for _, r := range existRoutes {
Expand All @@ -799,11 +805,39 @@ func routeDiff(existRoutes []netlink.Route, cidrs []string) (toAdd, toDel []stri
}
}
if !found {
toAdd = append(toAdd, c)
var priority int
var src, gw net.IP
scope := netlink.SCOPE_UNIVERSE
proto := netlink.RouteProtocol(syscall.RTPROT_STATIC)
if slices.Contains(joinCIDR, c) {
if util.CheckProtocol(c) == kubeovnv1.ProtocolIPv4 {
src = net.ParseIP(joinIPv4)
} else {
priority = 256
}
scope = netlink.SCOPE_LINK
proto = netlink.RouteProtocol(unix.RTPROT_KERNEL)
} else {
switch util.CheckProtocol(c) {
case kubeovnv1.ProtocolIPv4:
gw = gwV4
case kubeovnv1.ProtocolIPv6:
gw = gwV6
}
}
_, cidr, _ := net.ParseCIDR(c)
toAdd = append(toAdd, netlink.Route{
Dst: cidr,
Src: src,
Gw: gw,
Protocol: proto,
Scope: scope,
Priority: priority,
})
}
}
if len(toAdd) > 0 {
klog.Infof("route to add %v", toAdd)
klog.Infof("routes to add: %v", toAdd)
}
return
}
Expand Down Expand Up @@ -939,9 +973,8 @@ func (c *Controller) getPolicyRouting(subnet *kubeovnv1.Subnet) ([]netlink.Rule,
// routes
var routes []netlink.Route
for i := range protocols {
family, _ := util.ProtocolToFamily(protocols[i])
routes = append(routes, netlink.Route{
Protocol: netlink.RouteProtocol(family),
Protocol: netlink.RouteProtocol(syscall.RTPROT_STATIC),
Table: int(subnet.Spec.PolicyRoutingTableID),
Gw: net.ParseIP(egw[i]),
})
Expand Down
2 changes: 1 addition & 1 deletion pkg/daemon/gateway.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -271,7 +271,7 @@ func (c *Controller) deletePodPolicyRouting(podProtocol, externalEgressGateway s

func (c *Controller) addPolicyRouting(family int, gateway string, priority, tableID uint32, ips ...string) error {
route := &netlink.Route{
Protocol: netlink.RouteProtocol(family),
Protocol: netlink.RouteProtocol(syscall.RTPROT_STATIC),
Gw: net.ParseIP(gateway),
Table: int(tableID),
}
Expand Down
23 changes: 20 additions & 3 deletions pkg/daemon/ovs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -406,19 +406,36 @@ func configureNodeNic(portName, ip, gw, joinCIDR string, macAddr net.HardwareAdd
}
}
if !found {
protocol := util.CheckProtocol(c)
var src net.IP
var priority int
if protocol == kubeovnv1.ProtocolIPv4 {
for _, ip := range strings.Split(ipStr, ",") {
if util.CheckProtocol(ip) == protocol {
src = net.ParseIP(ip)
break
}
}
} else {
priority = 256
}
_, cidr, _ := net.ParseCIDR(c)
toAdd = append(toAdd, netlink.Route{
Dst: cidr,
Scope: netlink.SCOPE_UNIVERSE,
Dst: cidr,
Src: src,
Protocol: netlink.RouteProtocol(unix.RTPROT_KERNEL),
Scope: netlink.SCOPE_LINK,
Priority: priority,
})
}
}
if len(toAdd) > 0 {
klog.Infof("route to add for nic %s, %v", util.NodeNic, toAdd)
klog.Infof("routes to be added on nic %s: %v", util.NodeNic, toAdd)
}

for _, r := range toAdd {
r.LinkIndex = hostLink.Attrs().Index
klog.Infof("adding route %q on %s", r.String(), hostLink.Attrs().Name)
if err = netlink.RouteReplace(&r); err != nil && !errors.Is(err, syscall.EEXIST) {
klog.Errorf("failed to replace route %v: %v", r, err)
}
Expand Down

0 comments on commit 772c19c

Please sign in to comment.