Add OVS flows for implementing Egress

Add flows that set pkt_mark for egress traffic that should be SNAT'd
using a SNAT IP, including egress traffic from a local Pod to which the
Egress is applied, and traffic from a remote Node that is tunnelled to
the egress Node with the SNAT IP.
Each SNAT IP on the Node will be allocated with a unique integer ID,
which is set to the pkt_mark, and so the SNAT implementation can look
up the right SNAT IP from the pkt_mark. On Linux, SNAT will be
implemented by iptables SNAT rules; on Windows, SNAT is implemented
by OVS NAT.

go.mod

@@ -72,7 +72,7 @@ replace (
 	github.com/Microsoft/hcsshim v0.8.9 => github.com/ruicao93/hcsshim v0.8.10-0.20210114035434-63fe00c1b9aa
 	// antrea/plugins/octant/go.mod also has this replacement since replace statement in dependencies
 	// were ignored. We need to change antrea/plugins/octant/go.mod if there is any change here.
-	github.com/contiv/ofnet => github.com/wenyingd/ofnet v0.0.0-20210205051801-5a4f247248d4
+	github.com/contiv/ofnet => github.com/wenyingd/ofnet v0.0.0-20210318032909-171b6795a2da
 	// fake.NewSimpleClientset is quite slow when it's initialized with massive objects due to
 	// https://github.com/kubernetes/kubernetes/issues/89574. It takes more than tens of minutes to
 	// init a fake client with 200k objects, which makes it hard to run the NetworkPolicy scale test.

go.sum

@@ -409,8 +409,8 @@ github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df h1:OviZH7qLw/7Zo
 github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/vmware/go-ipfix v0.4.5 h1:EwG2bQXKT72IzMOsCcbvP1Po2PncLoSoPuYrHf3YrsI=
 github.com/vmware/go-ipfix v0.4.5/go.mod h1:lQz3f4r2pZWo0q8s8BtZ0xo5fPSOYsYteqJgBASP69o=
-github.com/wenyingd/ofnet v0.0.0-20210205051801-5a4f247248d4 h1:HwolNov6r/aM4zwA3MiSzxJKUTi3MypPOR6PRCTg1sA=
-github.com/wenyingd/ofnet v0.0.0-20210205051801-5a4f247248d4/go.mod h1:8mMMWAYBNUeTGXYKizOLETfN3WIbu3P5DgvS2jiXKdI=
+github.com/wenyingd/ofnet v0.0.0-20210318032909-171b6795a2da h1:ragN21nQa4zKuCwR2UEbTXEAh3L2YN/Id5SCVkjjwdY=
+github.com/wenyingd/ofnet v0.0.0-20210318032909-171b6795a2da/go.mod h1:8mMMWAYBNUeTGXYKizOLETfN3WIbu3P5DgvS2jiXKdI=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xlab/handysort v0.0.0-20150421192137-fb3537ed64a1/go.mod h1:QcJo0QPSfTONNIgpN5RA8prR7fF8nkF6cTWTcNerRO8=

pkg/agent/openflow/client.go

@@ -157,6 +157,25 @@ type Client interface {
 	// SNAT with the Openflow NAT action.
 	InstallExternalFlows() error
 
+	// InstallSNATMarkFlows installs flows for a local SNAT IP. On Linux, a
+	// single flow is added to mark the packets tunnelled from remote Nodes
+	// that should be SNAT'd with the SNAT IP. On Windows, an extra flow is
+	// added to perform SNAT for the marked packets with the SNAT IP.
+	InstallSNATMarkFlows(snatIP net.IP, mark uint32) error
+
+	// UninstallSNATMarkFlows removes the flows installed to set the packet
+	// mark for a SNAT IP.
+	UninstallSNATMarkFlows(mark uint32) error
+
+	// InstallSNATPolicyFlow installs the SNAT policy flows for a local Pod.
+	// which set the SNAT IP mark on the packets from the ofPort to external.
+	// As of now, a Pod can be configured with a single SNAT IP in a single
+	// address family (IPv4 or IPv6).
+	InstallPodSNATFlows(ofPort uint32, snatMark uint32, isIPv6 bool) error
+
+	// UninstallPodSNATFlows removes the SNAT policy flows for the local Pod.
+	UninstallPodSNATFlows(ofPort uint32) error
+
 	// Disconnect disconnects the connection between client and OFSwitch.
 	Disconnect() error
 
@@ -647,6 +666,41 @@ func (c *client) InstallExternalFlows() error {
 	return nil
 }
 
+func (c *client) InstallSNATMarkFlows(snatIP net.IP, mark uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	flows := c.snatMarkFlows(snatIP, mark)
+	cacheKey := fmt.Sprintf("s%x", mark)
+	return c.addFlows(c.snatFlowCache, cacheKey, flows)
+}
+
+func (c *client) UninstallSNATMarkFlows(mark uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	cacheKey := fmt.Sprintf("s%x", mark)
+	return c.deleteFlows(c.snatFlowCache, cacheKey)
+}
+
+func (c *client) InstallPodSNATFlows(ofPort uint32, snatMark uint32, isIPv6 bool) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	flows := make([]binding.Flow, 0, 2)
+	if !isIPv6 {
+		flows = append(flows, c.snatRuleFlow(ofPort, snatMark, binding.ProtocolIP))
+	} else {
+		flows = append(flows, c.snatRuleFlow(ofPort, snatMark, binding.ProtocolIPv6))
+	}
+	cacheKey := fmt.Sprintf("p%x", ofPort)
+	return c.addFlows(c.snatFlowCache, cacheKey, flows)
+}
+
+func (c *client) UninstallPodSNATFlows(ofPort uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	cacheKey := fmt.Sprintf("p%x", ofPort)
+	return c.deleteFlows(c.snatFlowCache, cacheKey)
+}
+
 func (c *client) ReplayFlows() {
 	c.replayMutex.Lock()
 	defer c.replayMutex.Unlock()

pkg/agent/openflow/client_windows.go

@@ -18,7 +18,6 @@
 package openflow
 
 import (
-	"fmt"
 	"net"
 
 	"github.com/vmware-tanzu/antrea/pkg/agent/openflow/cookie"
@@ -39,13 +38,13 @@ func (c *client) InstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPor
 	defer c.replayMutex.RUnlock()
 	var flows []binding.Flow
 	flows = append(flows, c.loadBalancerServiceFromOutsideFlow(svcIP, svcPort, protocol))
-	cacheKey := fmt.Sprintf("L%s%s%x", svcIP, protocol, svcPort)
+	cacheKey := fmt.Sprintf("LoadBalancerService_%s_%d_%s", svcIP, svcPort, protocol)
 	return c.addFlows(c.serviceFlowCache, cacheKey, flows)
 }
 
 func (c *client) UninstallLoadBalancerServiceFromOutsideFlows(svcIP net.IP, svcPort uint16, protocol binding.Protocol) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
-	cacheKey := fmt.Sprintf("L%s%s%x", svcIP, protocol, svcPort)
+	cacheKey := fmt.Sprintf("LoadBalancerService_%s_%d_%s", svcIP, svcPort, protocol)
 	return c.deleteFlows(c.serviceFlowCache, cacheKey)
 }

pkg/agent/openflow/pipeline.go

@@ -287,6 +287,10 @@ var (
 	// IPv4/v6 DSCP (bits 2-7) field supports exact match only.
 	traceflowTagToSRange = binding.Range{2, 7}
 
+	// snatPktMarkRange takes an 8-bit range of pkt_mark to store the ID of
+	// a SNAT IP. The bit range must match SNATIPMarkMask.
+	snatPktMarkRange = binding.Range{0, 7}
+
 	globalVirtualMAC, _ = net.ParseMAC("aa:bb:cc:dd:ee:ff")
 	hairpinIP           = net.ParseIP("169.254.169.252").To4()
 	hairpinIPv6         = net.ParseIP("fc00::aabb:ccdd:eeff").To16()
@@ -317,16 +321,17 @@ func portToUint16(port int) uint16 {
 }
 
 type client struct {
-	enableProxy                                   bool
-	enableAntreaPolicy                            bool
-	enableEgress                                  bool
-	roundInfo                                     types.RoundInfo
-	cookieAllocator                               cookie.Allocator
-	bridge                                        binding.Bridge
-	egressEntryTable                              binding.TableIDType
-	ingressEntryTable                             binding.TableIDType
-	pipeline                                      map[binding.TableIDType]binding.Table
-	nodeFlowCache, podFlowCache, serviceFlowCache *flowCategoryCache // cache for corresponding deletions
+	enableProxy        bool
+	enableAntreaPolicy bool
+	enableEgress       bool
+	roundInfo          types.RoundInfo
+	cookieAllocator    cookie.Allocator
+	bridge             binding.Bridge
+	egressEntryTable   binding.TableIDType
+	ingressEntryTable  binding.TableIDType
+	pipeline           map[binding.TableIDType]binding.Table
+	// Flow caches for corresponding deletions.
+	nodeFlowCache, podFlowCache, serviceFlowCache, snatFlowCache *flowCategoryCache
 	// "fixed" flows installed by the agent after initialization and which do not change during
 	// the lifetime of the client.
 	gatewayFlows, defaultServiceFlows, defaultTunnelFlows, hostNetworkingFlows []binding.Flow
@@ -1638,6 +1643,34 @@ func (c *client) snatCommonFlows(nodeIP net.IP, localSubnet net.IPNet, localGate
 	return flows
 }
 
+// snatIPFromTunnelFlow generates a flow that marks SNAT packets tunnelled from
+// remote Nodes. The SNAT IP matches the packet's tunnel destination IP.
+func (c *client) snatIPFromTunnelFlow(snatIP net.IP, mark uint32) binding.Flow {
+	ipProto := getIPProtocol(snatIP)
+	return c.pipeline[snatTable].BuildFlow(priorityNormal).
+		MatchProtocol(ipProto).
+		MatchCTStateNew(true).MatchCTStateTrk(true).
+		MatchTunnelDst(snatIP).
+		Action().LoadPktMarkRange(mark, snatPktMarkRange).
+		Action().GotoTable(l3DecTTLTable).
+		Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
+		Done()
+}
+
+// snatRuleFlow generates a flow that sets the packet mark with the ID of the
+// SNAT IP, for the traffic from the ofPort to external.
+func (c *client) snatRuleFlow(ofPort uint32, snatMark uint32, protocol binding.Protocol) binding.Flow {
+	snatTable := c.pipeline[snatTable]
+	return snatTable.BuildFlow(priorityNormal).
+		MatchProtocol(protocol).
+		MatchCTStateNew(true).MatchCTStateTrk(true).
+		MatchInPort(ofPort).
+		Action().LoadPktMarkRange(snatMark, snatPktMarkRange).
+		Action().GotoTable(snatTable.GetNext()).
+		Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
+		Done()
+}
+
 // loadBalancerServiceFromOutsideFlow generates the flow to forward LoadBalancer service traffic from outside node
 // to gateway. kube-proxy will then handle the traffic.
 // This flow is for Windows Node only.
@@ -1941,6 +1974,9 @@ func NewClient(bridgeName, mgmtAddr string, ovsDatapathType ovsconfig.OVSDatapat
 	} else {
 		c.egressEntryTable, c.ingressEntryTable = EgressRuleTable, IngressRuleTable
 	}
+	if enableEgress {
+		c.snatFlowCache = newFlowCategoryCache()
+	}
 	c.generatePipeline()
 	return c
 }

pkg/agent/openflow/pipeline_other.go

@@ -31,3 +31,7 @@ func (c *client) externalFlows(nodeIP net.IP, localSubnet net.IPNet, localGatewa
 	}
 	return c.snatCommonFlows(nodeIP, localSubnet, localGatewayMAC, cookie.SNAT)
 }
+
+func (c *client) snatMarkFlows(snatIP net.IP, mark uint32) []binding.Flow {
+	return []binding.Flow{c.snatIPFromTunnelFlow(snatIP, mark)}
+}

pkg/agent/openflow/pipeline_windows.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/vmware-tanzu/antrea/pkg/agent/config"
 	"github.com/vmware-tanzu/antrea/pkg/agent/openflow/cookie"
+	"github.com/vmware-tanzu/antrea/pkg/agent/types"
 	binding "github.com/vmware-tanzu/antrea/pkg/ovs/openflow"
 )
 
@@ -197,6 +198,38 @@ func (c *client) externalFlows(nodeIP net.IP, localSubnet net.IPNet, localGatewa
 	return flows
 }
 
+func (c *client) snatMarkFlows(snatIP net.IP, mark uint32) []binding.Flow {
+	ipProto := getIPProtocol(snatIP)
+	snatIPRange := &binding.IPRange{StartIP: snatIP, EndIP: snatIP}
+	ctCommitTable := c.pipeline[conntrackCommitTable]
+	nextTable := ctCommitTable.GetNext()
+	flows := []binding.Flow{
+		c.snatIPFromTunnelFlow(snatIP, mark),
+		ctCommitTable.BuildFlow(priorityNormal).
+			MatchProtocol(binding.ProtocolIP).
+			MatchCTStateNew(true).MatchCTStateTrk(true).MatchCTStateDNAT(false).
+			MatchPktMark(mark, &types.SNATIPMarkMask).
+			Action().CT(true, nextTable, CtZone).
+			SNAT(snatIPRange, nil).
+			LoadToMark(snatCTMark).CTDone().
+			Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
+			Done(),
+	}
+
+	if c.enableAntreaProxy {
+		flows = append(flows, ctCommitTable.BuildFlow(priorityNormal).
+			MatchProtocol(binding.ProtocolIP).
+			MatchCTStateNew(true).MatchCTStateTrk(true).MatchCTStateDNAT(true).
+			MatchPktMark(mark, &types.SNATIPMarkMask).
+			Action().CT(true, nextTable, ctZoneSNAT).
+			SNAT(snatIPRange, nil).
+			LoadToMark(snatCTMark).CTDone().
+			Cookie(c.cookieAllocator.Request(cookie.SNAT).Raw()).
+			Done())
+	}
+	return flows
+}
+
 // hostBridgeUplinkFlows generates the flows that forward traffic between the
 // bridge local port and the uplink port to support the host traffic with
 // outside.

pkg/agent/types/marks.go

@@ -23,4 +23,8 @@ const (
 var (
 	// HostLocalSourceMark is the mark generated from HostLocalSourceBit.
 	HostLocalSourceMark = uint32(1 << HostLocalSourceBit)
+
+	// SNATIPMarkMask is the bits of packet mark that stores the ID of the
+	// SNAT IP for a "Pod -> external" egress packet, that is to be SNAT'd.
+	SNATIPMarkMask = 0xFF
 )

pkg/ovs/openflow/interfaces.go

@@ -69,6 +69,7 @@ const (
 	NxmFieldTunMetadata = "NXM_NX_TUN_METADATA"
 	NxmFieldIPToS       = "NXM_OF_IP_TOS"
 	NxmFieldXXReg       = "NXM_NX_XXREG"
+	NxmFieldPktMark     = "NXM_NX_PKT_MARK"
 )
 
 const (
@@ -170,6 +171,7 @@ type Flow interface {
 type Action interface {
 	LoadARPOperation(value uint16) FlowBuilder
 	LoadRegRange(regID int, value uint32, to Range) FlowBuilder
+	LoadPktMarkRange(value uint32, to Range) FlowBuilder
 	LoadRange(name string, addr uint64, to Range) FlowBuilder
 	Move(from, to string) FlowBuilder
 	MoveRange(fromName, toName string, from, to Range) FlowBuilder
@@ -234,6 +236,7 @@ type FlowBuilder interface {
 	MatchDstPort(port uint16, portMask *uint16) FlowBuilder
 	MatchICMPv6Type(icmp6Type byte) FlowBuilder
 	MatchICMPv6Code(icmp6Code byte) FlowBuilder
+	MatchTunnelDst(dstIP net.IP) FlowBuilder
 	MatchTunMetadata(index int, data uint32) FlowBuilder
 	// MatchCTSrcIP matches the source IPv4 address of the connection tracker original direction tuple.
 	MatchCTSrcIP(ip net.IP) FlowBuilder

pkg/ovs/openflow/ofctrl_action.go

@@ -239,6 +239,11 @@ func (a *ofFlowAction) LoadRegRange(regID int, value uint32, rng Range) FlowBuil
 	return a.builder
 }
 
+// LoadToPktMarkRange is an action to load data into pkt_mark at specified range.
+func (a *ofFlowAction) LoadPktMarkRange(value uint32, rng Range) FlowBuilder {
+	return a.LoadRange(NxmFieldPktMark, uint64(value), rng)
+}
+
 // Move is an action to copy all data from "fromField" to "toField". Fields with name "fromField" and "fromField" should
 // have the same data length, otherwise there will be error when realizing the flow on OFSwitch.
 func (a *ofFlowAction) Move(fromField, toField string) FlowBuilder {

pkg/ovs/openflow/ofctrl_builder.go

@@ -251,6 +251,17 @@ func (b *ofFlowBuilder) MatchPktMark(value uint32, mask *uint32) FlowBuilder {
 	return b
 }
 
+// MatchTunnelDst adds match condition for matching tun_dst or tun_ipv6_dst.
+func (b *ofFlowBuilder) MatchTunnelDst(dstIP net.IP) FlowBuilder {
+	if dstIP.To4() != nil {
+		b.matchers = append(b.matchers, fmt.Sprintf("tun_dst=%s", dstIP.String()))
+	} else {
+		b.matchers = append(b.matchers, fmt.Sprintf("tun_ipv6_dst=%s", dstIP.String()))
+	}
+	b.ofFlow.Match.TunnelDst = &dstIP
+	return b
+}
+
 func ctLabelRange(high, low uint64, rng Range, match *ofctrl.FlowMatch) {
 	// [127..64] [63..0]
 	//   high     low