Add open flow sync refer to ovn-k8s (#6117)

* add openflow sync refer to ovn-k8s

Signed-off-by: clyi <clyi@alauda.io>

Author: changluyi

pkg/daemon/controller.go

@@ -706,6 +706,9 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 		}
 	}
 
+	// Start OpenFlow sync loop
+	go c.runFlowSync(stopCh)
+
 	<-stopCh
 	klog.Info("Shutting down workers")
 }

pkg/daemon/controller_linux.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 	"slices"
 	"strings"
+	"sync"
 	"syscall"
 
 	ovsutil "github.com/digitalocean/go-openvswitch/ovs"
@@ -49,6 +50,10 @@ type ControllerRuntime struct {
 
 	nmSyncer  *networkManagerSyncer
 	ovsClient *ovsutil.Client
+
+	flowCache      map[string]map[string][]string // key: bridgeName -> flowKey -> flow rules
+	flowCacheMutex sync.RWMutex
+	flowChan       chan struct{} // channel to trigger immediate flow sync
 }
 
 type LbServiceRules struct {
@@ -103,6 +108,10 @@ func (c *Controller) initRuntime() error {
 	c.k8sipsets = k8sipset.New(c.k8sExec)
 	c.ovsClient = ovsutil.New()
 
+	// Initialize OpenFlow flow cache (ovn-kubernetes style)
+	c.flowCache = make(map[string]map[string][]string)
+	c.flowChan = make(chan struct{}, 1)
+
 	if c.protocol == kubeovnv1.ProtocolIPv4 || c.protocol == kubeovnv1.ProtocolDual {
 		ipt, err := iptables.NewWithProtocol(iptables.ProtocolIPv4)
 		if err != nil {
@@ -493,18 +502,17 @@ func (c *Controller) reconcileServices(event *serviceEvent) error {
 	if len(lbServiceRulesToAdd) > 0 {
 		for _, rule := range lbServiceRulesToAdd {
 			klog.Infof("Adding LB service rule: %+v", rule)
-			if err := ovs.AddOrUpdateUnderlaySubnetSvcLocalOpenFlow(c.ovsClient, rule.BridgeName, rule.IP, rule.Protocol, rule.DstMac, rule.UnderlayNic, rule.Port); err != nil {
-				klog.Errorf("failed to add or update underlay subnet svc local openflow: %v", err)
+			if err := c.AddOrUpdateUnderlaySubnetSvcLocalFlowCache(rule.IP, rule.Port, rule.Protocol, rule.DstMac, rule.UnderlayNic, rule.BridgeName); err != nil {
+				klog.Errorf("failed to update underlay subnet svc local openflow cache: %v", err)
+				return err
 			}
 		}
 	}
 
 	if len(lbServiceRulesToDel) > 0 {
 		for _, rule := range lbServiceRulesToDel {
 			klog.Infof("Delete LB service rule: %+v", rule)
-			if err := ovs.DeleteUnderlaySubnetSvcLocalOpenFlow(c.ovsClient, rule.BridgeName, rule.IP, rule.Protocol, rule.UnderlayNic, rule.Port); err != nil {
-				klog.Errorf("failed to delete underlay subnet svc local openflow: %v", err)
-			}
+			c.deleteUnderlaySubnetSvcLocalFlowCache(rule.BridgeName, rule.IP, rule.Port, rule.Protocol)
 		}
 	}
 

pkg/daemon/flow_rules_linux.go

@@ -0,0 +1,89 @@
+package daemon
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"k8s.io/klog/v2"
+
+	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
+	"github.com/kubeovn/kube-ovn/pkg/util"
+)
+
+const flowKindUnderlayService = "usvc"
+
+func (c *Controller) AddOrUpdateUnderlaySubnetSvcLocalFlowCache(serviceIP string, port uint16, protocol, dstMac, underlayNic, bridgeName string) error {
+	inPort, err := c.getPortID(bridgeName, underlayNic)
+	if err != nil {
+		return err
+	}
+
+	outPort, err := c.getPortID(bridgeName, "patch-localnet.")
+	if err != nil {
+		return err
+	}
+
+	isIPv6 := util.CheckProtocol(serviceIP) == kubeovnv1.ProtocolIPv6
+	protoStr := ""
+	switch strings.ToUpper(protocol) {
+	case "TCP":
+		protoStr = "tcp"
+		if isIPv6 {
+			protoStr = "tcp6"
+		}
+	case "UDP":
+		protoStr = "udp"
+		if isIPv6 {
+			protoStr = "udp6"
+		}
+	default:
+		return fmt.Errorf("unsupported protocol %s", protocol)
+	}
+
+	cookie := fmt.Sprintf("0x%x", util.UnderlaySvcLocalOpenFlowCookieV4)
+	nwDst := "nw_dst"
+	if isIPv6 {
+		cookie = fmt.Sprintf("0x%x", util.UnderlaySvcLocalOpenFlowCookieV6)
+		nwDst = "ipv6_dst"
+	}
+
+	flow := fmt.Sprintf("cookie=%s,priority=%d,in_port=%d,%s,%s=%s,tp_dst=%d "+
+		"actions=mod_dl_dst:%s,output:%d",
+		cookie, util.UnderlaySvcLocalOpenFlowPriority, inPort, protoStr, nwDst, serviceIP, port, dstMac, outPort)
+
+	key := buildFlowKey(flowKindUnderlayService, serviceIP, port, protocol, "")
+	c.setFlowCache(c.flowCache, bridgeName, key, []string{flow})
+
+	klog.V(5).Infof("updated underlay flow cache for service %s", key)
+	c.requestFlowSync()
+	return nil
+}
+
+func (c *Controller) deleteUnderlaySubnetSvcLocalFlowCache(bridgeName, serviceIP string, port uint16, protocol string) {
+	key := buildFlowKey(flowKindUnderlayService, serviceIP, port, protocol, "")
+
+	c.deleteFlowCache(c.flowCache, bridgeName, key)
+
+	klog.V(5).Infof("deleted underlay flow cache for service %s", key)
+	c.requestFlowSync()
+}
+
+func buildFlowKey(kind, ip string, port uint16, protocol, extra string) string {
+	if extra == "" {
+		return fmt.Sprintf("%s-%s-%s-%d", kind, ip, protocol, port)
+	}
+	return fmt.Sprintf("%s-%s-%s-%d-%s", kind, ip, protocol, port, extra)
+}
+
+func (c *Controller) getPortID(bridgeName, portName string) (int, error) {
+	if c.ovsClient == nil {
+		return 0, errors.New("ovs client not initialized")
+	}
+
+	portInfo, err := c.ovsClient.OpenFlow.DumpPort(bridgeName, portName)
+	if err != nil {
+		return 0, fmt.Errorf("failed to dump port %s on bridge %s: %w", portName, bridgeName, err)
+	}
+	return portInfo.PortID, nil
+}

pkg/daemon/flow_sync_linux.go

@@ -0,0 +1,138 @@
+package daemon
+
+import (
+	"time"
+
+	openvswitch "github.com/digitalocean/go-openvswitch/ovs"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+
+	"github.com/kubeovn/kube-ovn/pkg/ovs"
+	"github.com/kubeovn/kube-ovn/pkg/util"
+)
+
+const flowSyncPeriod = 15 * time.Second
+
+var managedFlowCookieSet = sets.New[uint64](
+	util.UnderlaySvcLocalOpenFlowCookieV4,
+	util.UnderlaySvcLocalOpenFlowCookieV6,
+)
+
+func (c *Controller) requestFlowSync() {
+	if c.flowChan == nil {
+		util.LogFatalAndExit(nil, "flowChan is not initialized")
+	}
+
+	select {
+	case c.flowChan <- struct{}{}:
+		klog.V(5).Info("OpenFlow sync requested")
+	default:
+		klog.V(5).Info("OpenFlow sync already requested")
+	}
+}
+
+func (c *Controller) syncFlows() {
+	if c.ovsClient == nil {
+		util.LogFatalAndExit(nil, "ovsClient is not initialized")
+	}
+
+	flowCacheByBridge := c.storeFlowCache()
+
+	bridges, err := ovs.Bridges()
+	if err != nil {
+		klog.Errorf("failed to list bridges: %v", err)
+		return
+	}
+
+	for _, bridgeName := range bridges {
+		existing, err := ovs.DumpFlows(c.ovsClient, bridgeName)
+		if err != nil {
+			klog.Errorf("failed to dump flows for bridge %s: %v", bridgeName, err)
+			continue
+		}
+
+		preserved := filterUnmanagedFlows(existing)
+		cachedFlows := flowCacheByBridge[bridgeName]
+		finalFlows := append(preserved, cachedFlows...)
+
+		if err := ovs.ReplaceFlows(bridgeName, finalFlows); err != nil {
+			klog.Errorf("failed to replace flows for bridge %s: %v", bridgeName, err)
+			continue
+		}
+		if len(cachedFlows) == 0 {
+			klog.V(5).Infof("no cached flows for bridge %s", bridgeName)
+			continue
+		}
+		klog.V(3).Infof("synced %d cached flows on bridge %s", len(cachedFlows), bridgeName)
+	}
+}
+
+func (c *Controller) storeFlowCache() map[string][]string {
+	snapshot := make(map[string][]string)
+
+	c.flowCacheMutex.RLock()
+	defer c.flowCacheMutex.RUnlock()
+
+	for bridgeName, entries := range c.flowCache {
+		for _, flows := range entries {
+			snapshot[bridgeName] = append(snapshot[bridgeName], flows...)
+		}
+	}
+
+	return snapshot
+}
+
+func filterUnmanagedFlows(flows []string) []string {
+	filtered := make([]string, 0, len(flows))
+	for _, flow := range flows {
+		if isManagedFlow(flow) {
+			continue
+		}
+		filtered = append(filtered, flow)
+	}
+	return filtered
+}
+
+func isManagedFlow(flow string) bool {
+	var f openvswitch.Flow
+	if err := f.UnmarshalText([]byte(flow)); err != nil {
+		return false
+	}
+	return managedFlowCookieSet.Has(f.Cookie)
+}
+
+func (c *Controller) runFlowSync(stopCh <-chan struct{}) {
+	klog.Info("Starting OpenFlow sync loop")
+
+	ticker := time.NewTicker(flowSyncPeriod)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			c.syncFlows()
+		case <-c.flowChan:
+			klog.V(5).Info("Immediate OpenFlow sync triggered")
+			c.syncFlows()
+			ticker.Reset(flowSyncPeriod)
+		case <-stopCh:
+			klog.Info("Stopping OpenFlow sync loop")
+			return
+		}
+	}
+}
+
+func (c *Controller) setFlowCache(cache map[string]map[string][]string, bridgeName, key string, flows []string) {
+	c.flowCacheMutex.Lock()
+	defer c.flowCacheMutex.Unlock()
+
+	if cache[bridgeName] == nil {
+		cache[bridgeName] = make(map[string][]string)
+	}
+	cache[bridgeName][key] = flows
+}
+
+func (c *Controller) deleteFlowCache(cache map[string]map[string][]string, bridgeName, key string) {
+	c.flowCacheMutex.Lock()
+	defer c.flowCacheMutex.Unlock()
+	delete(cache[bridgeName], key)
+}