optimize node port-group check (#1514)

pkg/controller/config.go

@@ -60,8 +60,9 @@ type Configuration struct {
 	PodNamespace string
 	PodNicType   string
 
-	WorkerNum int
-	PprofPort int
+	WorkerNum       int
+	PprofPort       int
+	NodePgProbeTime int
 
 	NetworkType          string
 	DefaultProviderName  string
@@ -110,8 +111,9 @@ func ParseFlags() (*Configuration, error) {
 		argClusterTcpSessionLoadBalancer = pflag.String("cluster-tcp-session-loadbalancer", "cluster-tcp-session-loadbalancer", "The name for cluster tcp session loadbalancer")
 		argClusterUdpSessionLoadBalancer = pflag.String("cluster-udp-session-loadbalancer", "cluster-udp-session-loadbalancer", "The name for cluster udp session loadbalancer")
 
-		argWorkerNum = pflag.Int("worker-num", 3, "The parallelism of each worker")
-		argPprofPort = pflag.Int("pprof-port", 10660, "The port to get profiling data")
+		argWorkerNum       = pflag.Int("worker-num", 3, "The parallelism of each worker")
+		argPprofPort       = pflag.Int("pprof-port", 10660, "The port to get profiling data")
+		argNodePgProbeTime = pflag.Int("nodepg-probe-time", 1, "The probe interval for node port-group, the unit is minute")
 
 		argNetworkType          = pflag.String("network-type", util.NetworkTypeGeneve, "The ovn network type")
 		argDefaultProviderName  = pflag.String("default-provider-name", "provider", "The vlan or vxlan type default provider interface name")
@@ -189,6 +191,7 @@ func ParseFlags() (*Configuration, error) {
 		ExternalGatewayVlanID:         *argExternalGatewayVlanID,
 		EnableEcmp:                    *argEnableEcmp,
 		EnableKeepVmIP:                *argKeepVmIP,
+		NodePgProbeTime:               *argNodePgProbeTime,
 	}
 
 	if config.NetworkType == util.NetworkTypeVlan && config.DefaultHostInterface == "" {

pkg/controller/controller.go

@@ -728,7 +728,7 @@ func (c *Controller) startWorkers(stopCh <-chan struct{}) {
 	go wait.Until(c.CheckGatewayReady, 5*time.Second, stopCh)
 
 	if c.config.EnableNP {
-		go wait.Until(c.CheckNodePortGroup, 10*time.Second, stopCh)
+		go wait.Until(c.CheckNodePortGroup, time.Duration(c.config.NodePgProbeTime)*time.Minute, stopCh)
 	}
 
 	go wait.Until(c.syncVmLiveMigrationPort, 15*time.Second, stopCh)

pkg/controller/node.go

@@ -853,13 +853,7 @@ func (c *Controller) retryDelDupChassis(attempts int, sleep int, f func(node *v1
 	return nil
 }
 
-func (c *Controller) fetchPodsOnNode(nodeName string) ([]string, error) {
-	pods, err := c.podsLister.List(labels.Everything())
-	if err != nil {
-		klog.Errorf("failed to list pods, %v", err)
-		return nil, err
-	}
-
+func (c *Controller) fetchPodsOnNode(nodeName string, pods []*v1.Pod) ([]string, error) {
 	ports := make([]string, 0, len(pods))
 	for _, pod := range pods {
 		if !isPodAlive(pod) || pod.Spec.HostNetwork || pod.Spec.NodeName != nodeName || pod.Annotations[util.LogicalRouterAnnotation] != util.DefaultVpc {
@@ -886,37 +880,13 @@ func (c *Controller) fetchPodsOnNode(nodeName string) ([]string, error) {
 	return ports, nil
 }
 
-func (c *Controller) checkPodsChangedOnNode(pgName string, ports []string) (bool, error) {
-	pgPorts, err := c.ovnClient.ListPgPorts(pgName)
-	if err != nil {
-		klog.Errorf("failed to fetch ports for pg %v, %v", pgName, err)
-		return false, err
-	}
-
-	nameIdMap, idNameMap, err := c.ovnClient.ListLspForNodePortgroup()
-	if err != nil {
-		klog.Errorf("failed to list lsp info, %v", err)
-		return false, err
-	}
-
-	portIds := make([]string, 0, len(ports))
+func (c *Controller) checkPodsChangedOnNode(pgName string, nameIdMap map[string]string, pgPorts, ports []string) (bool, error) {
 	for _, port := range ports {
 		if portId, ok := nameIdMap[port]; ok {
-			portIds = append(portIds, portId)
-		}
-	}
-
-	for _, portId := range portIds {
-		if !util.IsStringIn(portId, pgPorts) {
-			klog.Infof("pod on node changed, new added port %v should add to node port group %v", idNameMap[portId], pgName)
-			return true, nil
-		}
-	}
-
-	for _, pgPort := range pgPorts {
-		if !util.IsStringIn(pgPort, portIds) {
-			klog.Infof("pod on node changed, can not find match pod for port %v in node port group %v", pgPort, pgName)
-			return true, nil
+			if !util.IsStringIn(portId, pgPorts) {
+				klog.Infof("pod on node changed, new added port %v should add to node port group %v", port, pgName)
+				return true, nil
+			}
 		}
 	}
 
@@ -942,22 +912,36 @@ func (c *Controller) checkAndUpdateNodePortGroup() error {
 		return err
 	}
 
+	pods, err := c.podsLister.List(labels.Everything())
+	if err != nil {
+		klog.Errorf("failed to list pods, %v", err)
+		return err
+	}
+
+	nameIdMap, _, err := c.ovnClient.ListLspForNodePortgroup()
+	if err != nil {
+		klog.Errorf("failed to list lsp info, %v", err)
+		return err
+	}
+
+	namePortsMap, err := c.ovnClient.ListPgPortsForNodePortgroup()
+	if err != nil {
+		klog.Errorf("failed to list port-group info, %v", err)
+		return err
+	}
+
 	for _, node := range nodes {
-		// ovn acl doesn't support address_set name with '-', so replace '-' by '.'
+		// The port-group should already created when add node
 		pgName := strings.Replace(node.Annotations[util.PortNameAnnotation], "-", ".", -1)
 		nodeIP := node.Annotations[util.IpAddressAnnotation]
-		if err := c.ovnClient.CreateNpPortGroup(pgName, "node", node.Name); err != nil {
-			klog.Errorf("failed to create port group %v for node %s, %v", pgName, node.Name, err)
-			return err
-		}
 
-		ports, err := c.fetchPodsOnNode(node.Name)
+		ports, err := c.fetchPodsOnNode(node.Name, pods)
 		if err != nil {
 			klog.Errorf("failed to fetch pods for node %v, %v", node.Name, err)
 			return err
 		}
 
-		changed, err := c.checkPodsChangedOnNode(pgName, ports)
+		changed, err := c.checkPodsChangedOnNode(pgName, nameIdMap, namePortsMap[pgName], ports)
 		if err != nil {
 			klog.Errorf("failed to check pod status for node %v, %v", node.Name, err)
 			continue

pkg/ovs/ovn-nbctl.go

@@ -1745,6 +1745,30 @@ func (c Client) ListLspForNodePortgroup() (map[string]string, map[string]string,
 	return nameIdMap, idNameMap, nil
 }
 
+func (c Client) ListPgPortsForNodePortgroup() (map[string][]string, error) {
+	output, err := c.ovnNbCommand("--data=bare", "--format=csv", "--no-heading", "--columns=name,ports", "list", "port_group")
+	if err != nil {
+		klog.Errorf("failed to list port_group, %v", err)
+		return nil, err
+	}
+	lines := strings.Split(output, "\n")
+	namePortsMap := make(map[string][]string, len(lines))
+	for _, l := range lines {
+		if len(strings.TrimSpace(l)) == 0 {
+			continue
+		}
+		parts := strings.Split(strings.TrimSpace(l), ",")
+		if len(parts) != 2 {
+			continue
+		}
+		name := strings.TrimSpace(parts[0])
+		ports := strings.Fields(parts[1])
+		namePortsMap[name] = ports
+	}
+
+	return namePortsMap, nil
+}
+
 func (c Client) SetPortsToPortGroup(portGroup string, portNames []string) error {
 	ovnArgs := []string{"clear", "port_group", portGroup, "ports"}
 	if len(portNames) > 0 {