Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[performance] in large-scale clusters, init node router policy too slow #4895

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
+1,084 −56
Open

[performance] in large-scale clusters, init node router policy too slow #4895

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
2924f7c
[performance] init node router policy too slow, in large clusters
cmdy Jan 2, 2025
cccdcd0
Merge branch 'master' into fix-init-node-router_policy-slow
cmdy Jan 2, 2025
714986f
[fix] uncomment
cmdy Jan 2, 2025
3c7bb83
[fix] code standards
cmdy Jan 2, 2025
ec1a93c
[fix] delete whitespace
cmdy Jan 2, 2025
7373176
[fix] empty logical_router_policy slice parameters
cmdy Jan 3, 2025
2c29d48
[fix] uncomment
cmdy Jan 3, 2025
8f367a1
[fix] variable name
cmdy Jan 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 160 additions & 0 deletions mocks/pkg/ovs/interface.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

122 changes: 66 additions & 56 deletions pkg/controller/init.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -775,81 +775,91 @@ func (c *Controller) syncVlanCR() error {
return nil
}

func (c *Controller) migrateNodeRoute(af int, node, ip, nexthop string) error {
// default vpc use static route in old version, migrate to policy route
var (
match = fmt.Sprintf("ip%d.dst == %s", af, ip)
action = kubeovnv1.PolicyRouteActionReroute
externalIDs = map[string]string{
"vendor": util.CniTypeName,
"node": node,
func (c *Controller) batchMigrateNodeRoute(nodes []*v1.Node) error {
start := time.Now()
addPolicies := make([]*kubeovnv1.PolicyRoute, 0)
delPolicies := make([]*kubeovnv1.PolicyRoute, 0)
staticRoutes := make([]*kubeovnv1.StaticRoute, 0)
externalIDsMap := make(map[string]map[string]string)
delAsNames := make([]string, 0)
for _, node := range nodes {
if node.Annotations[util.AllocatedAnnotation] != "true" {
continue
}
nodeName := node.Name
nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
joinAddrV4, joinAddrV6 := util.SplitStringIP(node.Annotations[util.IPAddressAnnotation])
if nodeIPv4 != "" && joinAddrV4 != "" {
buildNodeRoute(4, nodeName, joinAddrV4, nodeIPv4, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)
}
if nodeIPv6 != "" && joinAddrV6 != "" {
buildNodeRoute(6, nodeName, joinAddrV6, nodeIPv6, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)
}
)
klog.V(3).Infof("add policy route for router: %s, priority: %d, match %s, action %s, nexthop %s, externalID %v",
c.config.ClusterRouter, util.NodeRouterPolicyPriority, match, action, nexthop, externalIDs)
if err := c.addPolicyRouteToVpc(
c.config.ClusterRouter,
&kubeovnv1.PolicyRoute{
Priority: util.NodeRouterPolicyPriority,
Match: match,
Action: action,
NextHopIP: nexthop,
},
externalIDs,
); err != nil {
klog.Errorf("failed to add logical router policy for node %s: %v", node, err)
return err
}

if err := c.deleteStaticRouteFromVpc(
c.config.ClusterRouter,
util.MainRouteTable,
ip,
"",
kubeovnv1.PolicyDst,
); err != nil {
klog.Errorf("failed to delete obsolete static route for node %s: %v", node, err)
if err := c.batchAddPolicyRouteToVpc(c.config.ClusterRouter, addPolicies, externalIDsMap); err != nil {
klog.Errorf("failed to batch add logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
return err
}

asName := nodeUnderlayAddressSetName(node, af)
obsoleteMatch := fmt.Sprintf("ip%d.dst == %s && ip%d.src != $%s", af, ip, af, asName)
klog.V(3).Infof("delete policy route for router: %s, priority: %d, match %s", c.config.ClusterRouter, util.NodeRouterPolicyPriority, obsoleteMatch)
if err := c.deletePolicyRouteFromVpc(c.config.ClusterRouter, util.NodeRouterPolicyPriority, obsoleteMatch); err != nil {
klog.Errorf("failed to delete obsolete logical router policy for node %s: %v", node, err)
if err := c.batchDeleteStaticRouteFromVpc(c.config.ClusterRouter, staticRoutes); err != nil {
klog.Errorf("failed to batch delete obsolete logical router static route for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
return err
}

if err := c.OVNNbClient.DeleteAddressSet(asName); err != nil {
klog.Errorf("delete obsolete address set %s for node %s: %v", asName, node, err)
if err := c.batchDeletePolicyRouteFromVpc(c.config.ClusterRouter, delPolicies); err != nil {
klog.Errorf("failed to batch delete obsolete logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
return err
}
if err := c.OVNNbClient.BatchDeleteAddressSetByNames(delAsNames); err != nil {
klog.Errorf("failed to batch delete obsolete address set for asNames %v nodes %d: %v", delAsNames, len(nodes), err)
return err
}
klog.V(3).Infof("take to %v batch migrate node route for router: %s priority: %d add policy len: %d extrenalID len: %d del policy len: %d del address set len: %d",
time.Since(start), c.config.ClusterRouter, util.NodeRouterPolicyPriority, len(addPolicies), len(externalIDsMap), len(delPolicies), len(delAsNames))

return nil
}

func buildNodeRoute(af int, nodeName, nexthop, ip string, addPolicies, delPolicies *[]*kubeovnv1.PolicyRoute, staticRoutes *[]*kubeovnv1.StaticRoute, externalIDsMap map[string]map[string]string, delAsNames *[]string) {
var (
match = fmt.Sprintf("ip%d.dst == %s", af, ip)
action = kubeovnv1.PolicyRouteActionReroute
externalIDs = map[string]string{
"vendor": util.CniTypeName,
"node": nodeName,
}
)
*addPolicies = append(*addPolicies, &kubeovnv1.PolicyRoute{
Priority: util.NodeRouterPolicyPriority,
Match: match,
Action: action,
NextHopIP: nexthop,
})
externalIDsMap[buildExternalIDsMapKey(match, string(action), util.NodeRouterPolicyPriority)] = externalIDs
*staticRoutes = append(*staticRoutes, &kubeovnv1.StaticRoute{
Policy: kubeovnv1.PolicyDst,
RouteTable: util.MainRouteTable,
NextHopIP: "",
CIDR: ip,
})
asName := nodeUnderlayAddressSetName(nodeName, af)
obsoleteMatch := fmt.Sprintf("ip%d.dst == %s && ip%d.src != $%s", af, ip, af, asName)
*delPolicies = append(*delPolicies, &kubeovnv1.PolicyRoute{
Match: obsoleteMatch,
Priority: util.NodeRouterPolicyPriority,
})
*delAsNames = append(*delAsNames, asName)
}

func (c *Controller) syncNodeRoutes() error {
nodes, err := c.nodesLister.List(labels.Everything())
if err != nil {
klog.Errorf("failed to list nodes: %v", err)
return err
}
for _, node := range nodes {
if node.Annotations[util.AllocatedAnnotation] != "true" {
continue
}
nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
joinAddrV4, joinAddrV6 := util.SplitStringIP(node.Annotations[util.IPAddressAnnotation])
if nodeIPv4 != "" && joinAddrV4 != "" {
if err = c.migrateNodeRoute(4, node.Name, nodeIPv4, joinAddrV4); err != nil {
klog.Errorf("failed to migrate IPv4 route for node %s: %v", node.Name, err)
}
}
if nodeIPv6 != "" && joinAddrV6 != "" {
if err = c.migrateNodeRoute(6, node.Name, nodeIPv6, joinAddrV6); err != nil {
klog.Errorf("failed to migrate IPv6 route for node %s: %v", node.Name, err)
}
}

if err := c.batchMigrateNodeRoute(nodes); err != nil {
klog.Errorf("failed to batch migrate node routes: %v", err)
return err
}

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