add openflow sync refer to ovn-kubernetes

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,16 @@ 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)
 			}
 		}
 	}
 
 	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/controller_windows.go

@@ -25,6 +25,12 @@ func (c *Controller) initRuntime() error {
 	return nil
 }
 
+func (c *Controller) syncFlows() {}
+
+func (c *Controller) runFlowSync(stopCh <-chan struct{}) {
+	<-stopCh
+}
+
 func (c *Controller) reconcileServices(_ *serviceEvent) error {
 	return nil
 }

pkg/daemon/flow_rules.go

@@ -0,0 +1,76 @@
+package daemon
+
+import (
+	"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)
+}

pkg/daemon/flow_sync_linux.go

@@ -0,0 +1,204 @@
+package daemon
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"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 = map[uint64]struct{}{
+	util.UnderlaySvcLocalOpenFlowCookieV4: {},
+	util.UnderlaySvcLocalOpenFlowCookieV6: {},
+}
+
+func (c *Controller) requestFlowSync() {
+	if c.flowChan == nil {
+		return
+	}
+
+	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 {
+		return
+	}
+
+	flowsByBridge := 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)
+		managed := flowsByBridge[bridgeName]
+		merged := append(preserved, managed...)
+
+		if err := ovs.ReplaceFlows(bridgeName, merged); err != nil {
+			klog.Errorf("failed to replace flows for bridge %s: %v", bridgeName, err)
+			continue
+		}
+		if len(managed) == 0 {
+			klog.V(5).Infof("no managed flows for bridge %s", bridgeName)
+			continue
+		}
+		klog.V(3).Infof("synced %d managed flows on bridge %s", len(managed), bridgeName)
+	}
+}
+
+func (c *Controller) storeFlowCache() map[string][]string {
+	snapshot := make(map[string][]string)
+
+	c.flowCacheMutex.RLock()
+	defer c.flowCacheMutex.RUnlock()
+
+	appendFlowCache(snapshot, c.flowCache)
+
+	return snapshot
+}
+
+func appendFlowCache(dst map[string][]string, src map[string]map[string][]string) {
+	for bridgeName, entries := range src {
+		if len(entries) == 0 {
+			if _, ok := dst[bridgeName]; !ok {
+				dst[bridgeName] = nil
+			}
+			continue
+		}
+		for _, flows := range entries {
+			if len(flows) == 0 {
+				if _, ok := dst[bridgeName]; !ok {
+					dst[bridgeName] = nil
+				}
+				continue
+			}
+			dst[bridgeName] = append(dst[bridgeName], flows...)
+		}
+	}
+}
+
+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 {
+	cookie, ok := extractFlowCookie(flow)
+	if !ok {
+		return false
+	}
+	_, exists := managedFlowCookieSet[cookie]
+	return exists
+}
+
+func extractFlowCookie(flow string) (uint64, bool) {
+	idx := strings.Index(flow, "cookie=")
+	if idx == -1 {
+		return 0, false
+	}
+	cookieField := flow[idx+len("cookie="):]
+	if comma := strings.Index(cookieField, ","); comma != -1 {
+		cookieField = cookieField[:comma]
+	}
+	if slash := strings.Index(cookieField, "/"); slash != -1 {
+		cookieField = cookieField[:slash]
+	}
+	cookieField = strings.TrimSpace(cookieField)
+	if cookieField == "" {
+		return 0, false
+	}
+
+	cookie, err := parseHexUint64(cookieField)
+	if err != nil {
+		return 0, false
+	}
+	return cookie, true
+}
+
+func parseHexUint64(value string) (uint64, error) {
+	if !strings.HasPrefix(value, "0x") {
+		value = "0x" + value
+	}
+	return strconv.ParseUint(value, 0, 64)
+}
+
+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) getPortID(bridgeName, portName string) (int, error) {
+	if c.ovsClient == nil {
+		return 0, fmt.Errorf("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
+}
+
+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()
+
+	entries := cache[bridgeName]
+	if entries == nil {
+		return
+	}
+	delete(entries, key)
+}