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

add OpenFlowManager package to manager OpenFlow rules per-bridge #72

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

add OpenFlowManager package to manager OpenFlow rules per-bridge #72

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
5 changes: 5 additions & 0 deletions go-controller/cmd/ovnkube/ovnkube.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/metrics"
controllerManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/network-controller-manager"
ovnnode "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/node"
OFManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/openflow-manager"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"

Expand Down Expand Up @@ -501,6 +502,10 @@ func runOvnKube(ctx context.Context, runMode *ovnkubeRunMode, ovnClientset *util
}
// register ovnkube node specific prometheus metrics exported by the node
metrics.RegisterNodeMetrics()

// initialization the global open flow manager
OFManager.NewOpenFlowCacheManager(wg, stopChan)

ncm, err := controllerManager.NewNodeNetworkControllerManager(ovnClientset, nodeWatchFactory, runMode.identity, eventRecorder)
if err != nil {
return fmt.Errorf("failed to create ovnkube node network controller manager: %w", err)
Expand Down
4 changes: 4 additions & 0 deletions go-controller/pkg/node/gateway_init_linux_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
adminpolicybasedrouteclient "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/adminpolicybasedroute/v1/apis/clientset/versioned/fake"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/kube"
OFManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/openflow-manager"
ovntest "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/testing"
linkMock "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/testing/mocks/github.com/vishvananda/netlink"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
Expand Down Expand Up @@ -265,6 +266,7 @@ func shareGatewayInterfaceTest(app *cli.App, testNS ns.NetNS,

gatewayNextHops, gatewayIntf, err := getGatewayNextHops()
Expect(err).NotTo(HaveOccurred())
OFManager.NewOpenFlowCacheManager(wg, stop)
ifAddrs := ovntest.MustParseIPNets(eth0CIDR)
sharedGw, err := newSharedGateway(nodeName, ovntest.MustParseIPNets(nodeSubnet), gatewayNextHops, gatewayIntf, "", ifAddrs, nodeAnnotator, k,
&fakeMgmtPortConfig, wf, rm)
Expand Down Expand Up @@ -629,6 +631,7 @@ func shareGatewayInterfaceDPUTest(app *cli.App, testNS ns.NetNS,

gatewayNextHops, gatewayIntf, err := getGatewayNextHops()
Expect(err).NotTo(HaveOccurred())
OFManager.NewOpenFlowCacheManager(wg, stop)
sharedGw, err := newSharedGateway(nodeName, ovntest.MustParseIPNets(nodeSubnet), gatewayNextHops,
gatewayIntf, "", ifAddrs, nodeAnnotator, k, &fakeMgmtPortConfig, wf, rm)
Expect(err).NotTo(HaveOccurred())
Expand Down Expand Up @@ -1058,6 +1061,7 @@ OFPT_GET_CONFIG_REPLY (xid=0x4): frags=normal miss_send_len=0`,

gatewayNextHops, gatewayIntf, err := getGatewayNextHops()
Expect(err).NotTo(HaveOccurred())
OFManager.NewOpenFlowCacheManager(wg, stop)
ifAddrs := ovntest.MustParseIPNets(eth0CIDR)
localGw, err := newLocalGateway(nodeName, ovntest.MustParseIPNets(nodeSubnet), gatewayNextHops, gatewayIntf, "", ifAddrs,
nodeAnnotator, &fakeMgmtPortConfig, k, wf, rm)
Expand Down
50 changes: 27 additions & 23 deletions go-controller/pkg/node/gateway_localnet_linux_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/config"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/kube"
OFManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/openflow-manager"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/retry"
ovntest "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/testing"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
Expand Down Expand Up @@ -47,15 +48,17 @@ func initFakeNodePortWatcher(iptV4, iptV6 util.IPTablesHelper) *nodePortWatcher
f6 := iptV6.(*util.FakeIPTables)
err = f6.MatchState(initIPTable)
Expect(err).NotTo(HaveOccurred())
bridgeName := "brp0"

id := OFManager.OpenFlowCacheManager.CreateFlowCache(bridgeName, false)
fNPW := nodePortWatcher{
ofportPhys: "eth0",
ofportPatch: "patch-breth0_ov",
gatewayIPv4: v4localnetGatewayIP,
gatewayIPv6: v6localnetGatewayIP,
serviceInfo: make(map[k8stypes.NamespacedName]*serviceConfig),
ofm: &openflowManager{
flowCache: map[string][]string{},
defaultBridgeFlowID: id,
},
}
return &fNPW
Expand Down Expand Up @@ -257,6 +260,7 @@ var _ = Describe("Node Operations", func() {
ipv4: &fakeMgmtPortIPFamilyConfig,
ipv6: nil,
}
OFManager.NewOpenFlowCacheManager(&sync.WaitGroup{}, make(chan struct{}))
fNPW = initFakeNodePortWatcher(iptV4, iptV6)
})

Expand Down Expand Up @@ -614,7 +618,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())
return nil
}
Expand Down Expand Up @@ -820,11 +824,11 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())
flows = fNPW.ofm.flowCache["Ingress_namespace1_service1_5.5.5.5_8080"]
flows = fNPW.ofm.getFlowCacheEntry("Ingress_namespace1_service1_5.5.5.5_8080")
Expect(flows).To(Equal(expectedLBIngressFlows))
flows = fNPW.ofm.flowCache["External_namespace1_service1_1.1.1.1_8080"]
flows = fNPW.ofm.getFlowCacheEntry("External_namespace1_service1_1.1.1.1_8080")
Expect(flows).To(Equal(expectedLBExternalIPFlows))

return nil
Expand Down Expand Up @@ -954,8 +958,8 @@ var _ = Describe("Node Operations", func() {

f4 := iptV4.(*util.FakeIPTables)
Expect(f4.MatchState(expectedTables)).To(Succeed())
Expect(fNPW.ofm.flowCache["Ingress_namespace1_service1_5.5.5.5_80"]).To(Equal(expectedLBIngressFlows))
Expect(fNPW.ofm.flowCache["External_namespace1_service1_1.1.1.1_80"]).To(Equal(expectedLBExternalIPFlows))
Expect(fNPW.ofm.getFlowCacheEntry("Ingress_namespace1_service1_5.5.5.5_80")).To(Equal(expectedLBIngressFlows))
Expect(fNPW.ofm.getFlowCacheEntry("External_namespace1_service1_1.1.1.1_80")).To(Equal(expectedLBExternalIPFlows))
return nil
}
Expect(app.Run([]string{app.Name})).To(Succeed())
Expand Down Expand Up @@ -1060,11 +1064,11 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())
flows = fNPW.ofm.flowCache["Ingress_namespace1_service1_5.5.5.5_8080"]
flows = fNPW.ofm.getFlowCacheEntry("Ingress_namespace1_service1_5.5.5.5_8080")
Expect(flows).To(Equal(expectedLBIngressFlows))
flows = fNPW.ofm.flowCache["External_namespace1_service1_1.1.1.1_8080"]
flows = fNPW.ofm.getFlowCacheEntry("External_namespace1_service1_1.1.1.1_8080")
Expect(flows).To(Equal(expectedLBExternalIPFlows))

return nil
Expand Down Expand Up @@ -1184,11 +1188,11 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(Equal(expectedNodePortFlows))
flows = fNPW.ofm.flowCache["Ingress_namespace1_service1_5.5.5.5_8080"]
flows = fNPW.ofm.getFlowCacheEntry("Ingress_namespace1_service1_5.5.5.5_8080")
Expect(flows).To(Equal(expectedLBIngressFlows))
flows = fNPW.ofm.flowCache["External_namespace1_service1_1.1.1.1_8080"]
flows = fNPW.ofm.getFlowCacheEntry("External_namespace1_service1_1.1.1.1_8080")
Expect(flows).To(Equal(expectedLBExternalIPFlows))

return nil
Expand Down Expand Up @@ -1993,7 +1997,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

addConntrackMocks(netlinkMock, []ctFilterDesc{{"10.129.0.2", 8080}, {"192.168.18.15", 31111}})
Expand Down Expand Up @@ -2035,7 +2039,7 @@ var _ = Describe("Node Operations", func() {
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())

flows = fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows = fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

return nil
Expand Down Expand Up @@ -2135,7 +2139,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(Equal(expectedFlows))

addConntrackMocks(netlinkMock, []ctFilterDesc{{"10.129.0.2", 8080}, {"192.168.18.15", 31111}})
Expand Down Expand Up @@ -2177,7 +2181,7 @@ var _ = Describe("Node Operations", func() {
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())

flows = fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows = fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

return nil
Expand Down Expand Up @@ -2282,7 +2286,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(Equal(expectedFlows))

addConntrackMocks(netlinkMock, []ctFilterDesc{{"10.129.0.2", 8080}, {"192.168.18.15", 31111}})
Expand Down Expand Up @@ -2324,7 +2328,7 @@ var _ = Describe("Node Operations", func() {
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())

flows = fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows = fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

return nil
Expand Down Expand Up @@ -2425,7 +2429,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(Equal(expectedFlows))

addConntrackMocks(netlinkMock, []ctFilterDesc{{"10.129.0.2", 8080}, {"192.168.18.15", 31111}})
Expand Down Expand Up @@ -2467,7 +2471,7 @@ var _ = Describe("Node Operations", func() {
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())

flows = fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows = fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

return nil
Expand Down Expand Up @@ -2571,7 +2575,7 @@ var _ = Describe("Node Operations", func() {
f4 := iptV4.(*util.FakeIPTables)
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())
flows := fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows := fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(Equal(expectedFlows))

addConntrackMocks(netlinkMock, []ctFilterDesc{{"10.129.0.2", 8080}, {"192.168.18.15", 31111}})
Expand Down Expand Up @@ -2613,7 +2617,7 @@ var _ = Describe("Node Operations", func() {
err = f4.MatchState(expectedTables)
Expect(err).NotTo(HaveOccurred())

flows = fNPW.ofm.flowCache["NodePort_namespace1_service1_tcp_31111"]
flows = fNPW.ofm.getFlowCacheEntry("NodePort_namespace1_service1_tcp_31111")
Expect(flows).To(BeNil())

return nil
Expand Down
14 changes: 9 additions & 5 deletions go-controller/pkg/node/gateway_shared_intf.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/kube"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/node/controllers/egressservice"
nodeipt "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/node/iptables"
OFManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/openflow-manager"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
"github.com/vishvananda/netlink"
Expand Down Expand Up @@ -1025,14 +1026,17 @@ func (npwipt *nodePortWatcherIptables) SyncServices(services []interface{}) erro
// -- to handle external -> service(ExternalTrafficPolicy: Local) -> host access without SNAT
func newGatewayOpenFlowManager(gwBridge, exGWBridge *bridgeConfiguration, subnets []*net.IPNet, extraIPs []net.IP) (*openflowManager, error) {
// add health check function to check default OpenFlow flows are on the shared gateway bridge
klog.Infof("Cathy newGatewayOpenFlowManager gwBridge %+v exGWBridge %+v", gwBridge, exGWBridge)
var dftID, extID string
dftID = OFManager.OpenFlowCacheManager.CreateFlowCache(gwBridge.bridgeName, true)
if exGWBridge != nil {
extID = OFManager.OpenFlowCacheManager.CreateFlowCache(exGWBridge.bridgeName, true)
}
ofm := &openflowManager{
defaultBridge: gwBridge,
externalGatewayBridge: exGWBridge,
flowCache: make(map[string][]string),
flowMutex: sync.Mutex{},
exGWFlowCache: make(map[string][]string),
exGWFlowMutex: sync.Mutex{},
flowChan: make(chan struct{}, 1),
defaultBridgeFlowID: dftID,
extGWBridgeFlowID: extID,
}

if err := ofm.updateBridgeFlowCache(subnets, extraIPs); err != nil {
Expand Down
70 changes: 12 additions & 58 deletions go-controller/pkg/node/openflow_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"sync"
"time"

OFManager "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/openflow-manager"
"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
"github.com/pkg/errors"

Expand All @@ -14,73 +15,30 @@ import (
type openflowManager struct {
defaultBridge *bridgeConfiguration
externalGatewayBridge *bridgeConfiguration
// flow cache, use map instead of array for readability when debugging
flowCache map[string][]string
flowMutex sync.Mutex
exGWFlowCache map[string][]string
exGWFlowMutex sync.Mutex
// channel to indicate we need to update flows immediately
flowChan chan struct{}
defaultBridgeFlowID string
extGWBridgeFlowID string
}

func (c *openflowManager) updateFlowCacheEntry(key string, flows []string) {
c.flowMutex.Lock()
defer c.flowMutex.Unlock()
c.flowCache[key] = flows
OFManager.OpenFlowCacheManager.UpdateFlowCacheEntry(c.defaultBridgeFlowID, key, flows, false)
}

func (c *openflowManager) deleteFlowsByKey(key string) {
c.flowMutex.Lock()
defer c.flowMutex.Unlock()
delete(c.flowCache, key)
OFManager.OpenFlowCacheManager.DeleteFlowsByKey(c.defaultBridgeFlowID, key, false)
}

func (c *openflowManager) updateExBridgeFlowCacheEntry(key string, flows []string) {
c.exGWFlowMutex.Lock()
defer c.exGWFlowMutex.Unlock()
c.exGWFlowCache[key] = flows
func (c *openflowManager) getFlowCacheEntry(key string) []string {
return OFManager.OpenFlowCacheManager.GetFlowsByKey(c.defaultBridgeFlowID, key)
}

func (c *openflowManager) requestFlowSync() {
select {
case c.flowChan <- struct{}{}:
klog.V(5).Infof("Gateway OpenFlow sync requested")
default:
klog.V(5).Infof("Gateway OpenFlow sync already requested")
}
func (c *openflowManager) updateExBridgeFlowCacheEntry(key string, flows []string) {
OFManager.OpenFlowCacheManager.UpdateFlowCacheEntry(c.extGWBridgeFlowID, key, flows, false)
}

func (c *openflowManager) syncFlows() {
// protect gwBridge config from being updated by gw.nodeIPManager
c.defaultBridge.Lock()
defer c.defaultBridge.Unlock()

c.flowMutex.Lock()
defer c.flowMutex.Unlock()

flows := []string{}
for _, entry := range c.flowCache {
flows = append(flows, entry...)
}

_, stderr, err := util.ReplaceOFFlows(c.defaultBridge.bridgeName, flows)
if err != nil {
klog.Errorf("Failed to add flows, error: %v, stderr, %s, flows: %s", err, stderr, c.flowCache)
}

func (c *openflowManager) requestFlowSync() {
OFManager.OpenFlowCacheManager.RequestFlowSync(c.defaultBridgeFlowID)
if c.externalGatewayBridge != nil {
c.exGWFlowMutex.Lock()
defer c.exGWFlowMutex.Unlock()

flows := []string{}
for _, entry := range c.exGWFlowCache {
flows = append(flows, entry...)
}

_, stderr, err := util.ReplaceOFFlows(c.externalGatewayBridge.bridgeName, flows)
if err != nil {
klog.Errorf("Failed to add flows, error: %v, stderr, %s, flows: %s", err, stderr, c.exGWFlowCache)
}
OFManager.OpenFlowCacheManager.RequestFlowSync(c.extGWBridgeFlowID)
}
}

Expand Down Expand Up @@ -109,10 +67,6 @@ func (c *openflowManager) Run(stopChan <-chan struct{}, doneWg *sync.WaitGroup)
continue
}
}
c.syncFlows()
case <-c.flowChan:
c.syncFlows()
timer.Reset(syncPeriod)
case <-stopChan:
return
}
Expand Down
Loading