Implement traffic control interfaces via OVS

Signed-off-by: Hongliang Liu <lhongliang@vmware.com>

cmd/antrea-agent/agent.go

@@ -130,7 +130,10 @@ func run(o *Options) error {
 		features.DefaultFeatureGate.Enabled(features.FlowExporter),
 		o.config.AntreaProxy.ProxyAll,
 		connectUplinkToBridge,
-		features.DefaultFeatureGate.Enabled(features.Multicast))
+		features.DefaultFeatureGate.Enabled(features.Multicast),
+		// TODO: replace with features.DefaultFeatureGate.Enabled(features.TrafficControl),
+		false,
+	)
 
 	_, serviceCIDRNet, _ := net.ParseCIDR(o.config.ServiceCIDR)
 	var serviceCIDRNetv6 *net.IPNet

pkg/agent/openflow/client.go

@@ -28,6 +28,7 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
 	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
+	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	utilip "antrea.io/antrea/pkg/util/ip"
 	"antrea.io/antrea/third_party/proxy"
@@ -279,6 +280,18 @@ type Client interface {
 		dstIP net.IP,
 		outPort uint32,
 		igmp ofutil.Message) error
+
+	// InstallTrafficControlMarkFlows installs the flows to mark the packets for a traffic control rule with the given mark value.
+	InstallTrafficControlMarkFlows(name string, mark uint32, srcOfPorts []uint32, direction v1alpha2.Direction) error
+
+	// UninstallTrafficControlMarkFlows removes the flows for a traffic control rule.
+	UninstallTrafficControlMarkFlows(name string) error
+
+	// InstallTrafficControlInitFlows installs the init flows for the given mark value.
+	InstallTrafficControlInitFlows(mark uint32, action v1alpha2.Action, dstOfInPort, dstOfOutPort uint32) error
+
+	// UninstallTrafficControlInitFlows removes the flows for the given mark value.
+	UninstallTrafficControlInitFlows(mark uint8) error
 }
 
 // GetFlowTableStatus returns an array of flow table status.
@@ -697,7 +710,8 @@ func (c *client) generatePipelines() {
 		c.networkConfig,
 		c.ovsDatapathType,
 		c.connectUplinkToBridge,
-		c.enableMulticast)
+		c.enableMulticast,
+		c.enableTrafficControl)
 	c.activatedFeatures = append(c.activatedFeatures, c.featurePodConnectivity)
 	c.traceableFeatures = append(c.traceableFeatures, c.featurePodConnectivity)
 
@@ -718,7 +732,8 @@ func (c *client) generatePipelines() {
 		c.bridge,
 		c.enableProxy,
 		c.proxyAll,
-		c.connectUplinkToBridge)
+		c.connectUplinkToBridge,
+		c.enableTrafficControl)
 	c.activatedFeatures = append(c.activatedFeatures, c.featureService)
 	c.traceableFeatures = append(c.traceableFeatures, c.featureService)
 
@@ -1128,3 +1143,35 @@ func (c *client) SendIGMPQueryPacketOut(
 	packetOutObj := packetOutBuilder.Done()
 	return c.bridge.SendPacketOut(packetOutObj)
 }
+
+func (c *client) InstallTrafficControlMarkFlows(name string, mark uint32, srcOfPorts []uint32, direction v1alpha2.Direction) error {
+	flows := c.featurePodConnectivity.trafficControlMarkFlows(mark, srcOfPorts, direction)
+	cacheKey := fmt.Sprintf("trafficControl_%s", name)
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	return c.addFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey, flows)
+}
+
+func (c *client) UninstallTrafficControlMarkFlows(name string) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	cacheKey := fmt.Sprintf("trafficControl_%s", name)
+	return c.deleteFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey)
+}
+
+func (c *client) InstallTrafficControlInitFlows(mark uint32, action v1alpha2.Action, dstOfInPort, dstOfOutPort uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	cacheKey := fmt.Sprintf("trafficControl_%d", mark)
+	flows := c.featurePodConnectivity.trafficControlInitFlows(mark, action, dstOfInPort, dstOfOutPort)
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	return c.addFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey, flows)
+}
+
+func (c *client) UninstallTrafficControlInitFlows(mark uint8) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+	cacheKey := fmt.Sprintf("trafficControl_%d", mark)
+	return c.deleteFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey)
+}

pkg/agent/openflow/client_test.go

@@ -107,7 +107,7 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -140,7 +140,7 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -186,7 +186,7 @@ func TestFlowInstallationFailed(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -225,7 +225,7 @@ func TestConcurrentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false, false, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -417,7 +417,7 @@ func Test_client_SendTraceflowPacket(t *testing.T) {
 }
 
 func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false, false)
 	c := ofClient.(*client)
 	c.cookieAllocator = cookie.NewAllocator(0)
 	c.nodeConfig = nodeConfig
@@ -443,7 +443,7 @@ func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
 }
 
 func prepareSendTraceflowPacket(ctrl *gomock.Controller, success bool) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false, false)
 	c := ofClient.(*client)
 	c.nodeConfig = nodeConfig
 	m := ovsoftest.NewMockBridge(ctrl)
@@ -531,7 +531,7 @@ func Test_client_setBasePacketOutBuilder(t *testing.T) {
 }
 
 func prepareSetBasePacketOutBuilder(ctrl *gomock.Controller, success bool) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false, false, false, false, false)
 	c := ofClient.(*client)
 	m := ovsoftest.NewMockBridge(ctrl)
 	c.bridge = m

pkg/agent/openflow/cookie/allocator.go

@@ -37,6 +37,7 @@ const (
 	Egress
 	Multicast
 	Traceflow
+	TrafficControl
 )
 
 func (c Category) String() string {

pkg/agent/openflow/fields.go

@@ -21,11 +21,12 @@ import (
 
 // Fields using reg.
 var (
-	tunnelVal  = uint32(1)
-	gatewayVal = uint32(2)
-	localVal   = uint32(3)
-	uplinkVal  = uint32(4)
-	bridgeVal  = uint32(5)
+	tunnelVal   = uint32(1)
+	gatewayVal  = uint32(2)
+	localVal    = uint32(3)
+	uplinkVal   = uint32(4)
+	bridgeVal   = uint32(5)
+	redirectVal = uint32(6)
 
 	// reg0 (NXM_NX_REG0)
 	// reg0[0..3]: Field to store the packet source. Marks in this field include:
@@ -34,12 +35,14 @@ var (
 	//   - 3: from local Pods.
 	//   - 4: from uplink port.
 	//   - 5: from bridge local port.
-	PktSourceField     = binding.NewRegField(0, 0, 3, "PacketSource")
-	FromTunnelRegMark  = binding.NewRegMark(PktSourceField, tunnelVal)
-	FromGatewayRegMark = binding.NewRegMark(PktSourceField, gatewayVal)
-	FromLocalRegMark   = binding.NewRegMark(PktSourceField, localVal)
-	FromUplinkRegMark  = binding.NewRegMark(PktSourceField, uplinkVal)
-	FromBridgeRegMark  = binding.NewRegMark(PktSourceField, bridgeVal)
+	//   - 6: from traffic control redirection out port.
+	PktSourceField      = binding.NewRegField(0, 0, 3, "PacketSource")
+	FromTunnelRegMark   = binding.NewRegMark(PktSourceField, tunnelVal)
+	FromGatewayRegMark  = binding.NewRegMark(PktSourceField, gatewayVal)
+	FromLocalRegMark    = binding.NewRegMark(PktSourceField, localVal)
+	FromUplinkRegMark   = binding.NewRegMark(PktSourceField, uplinkVal)
+	FromBridgeRegMark   = binding.NewRegMark(PktSourceField, bridgeVal)
+	FromRedirectRegMark = binding.NewRegMark(PktSourceField, redirectVal)
 	// reg0[4..7]: Field to store the packet destination. Marks in this field include:
 	//   - 1: to tunnel port.
 	//   - 2: to Antrea gateway port.
@@ -82,6 +85,8 @@ var (
 	CustomReasonDenyRegMark    = binding.NewRegMark(CustomReasonField, CustomReasonDeny)
 	CustomReasonDNSRegMark     = binding.NewRegMark(CustomReasonField, CustomReasonDNS)
 	CustomReasonIGMPRegMark    = binding.NewRegMark(CustomReasonField, CustomReasonIGMP)
+	// reg0[17..18]: Field to store the mark value for the packets that require mirorring or redirection.
+	TrafficControlField = binding.NewRegField(0, 18, 21, "TrafficControl")
 
 	// reg1(NXM_NX_REG1)
 	// Field to cache the ofPort of the OVS interface where to output packet.
@@ -175,7 +180,8 @@ var (
 
 	// CTMark[4]: Mark to indicate DNAT is performed on the connection for Service.
 	// This CT mark is used in CtZone / CtZoneV6 and SNATCtZone / SNATCtZoneV6.
-	ServiceCTMark = binding.NewOneBitCTMark(4)
+	ServiceCTMark    = binding.NewOneBitCTMark(4)
+	NotServiceCTMark = binding.NewOneBitZeroCTMark(4)
 
 	// CTMark[5]: Mark to indicate SNAT should be performed on the connection for Service.
 	// This CT mark is only used in CtZone / CtZoneV6.
@@ -184,6 +190,10 @@ var (
 	// CTMark[6]: Mark to indicate the connection is hairpin.
 	// This CT mark is used in CtZone / CtZoneV6 and SNATCtZone / SNATCtZoneV6.
 	HairpinCTMark = binding.NewOneBitCTMark(6)
+
+	// CTMark[6]: Field to store the mark value for connections that require mirorring or redirection.
+	// This CT mark is only used in CtZone / CtZoneV6.
+	ConnTrafficControlCTMarkField = binding.NewCTMarkField(7, 10)
 )
 
 // Fields using CT label.

pkg/agent/openflow/framework.go

@@ -187,6 +187,13 @@ func (f *featurePodConnectivity) getRequiredTables() []*Table {
 			}
 		}
 	}
+	if f.enableTrafficControl {
+		tables = append(tables,
+			EgressTrafficControlMarkTable,
+			TrafficControlRedirectOutTable,
+			IngressTrafficControlMarkTable,
+		)
+	}
 
 	return tables
 }

pkg/agent/openflow/network_policy_test.go

@@ -519,7 +519,7 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			mockOperations := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, false, true, false, false, false, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, false, true, false, false, false, false, false, false)
 			c = ofClient.(*client)
 			c.cookieAllocator = cookie.NewAllocator(0)
 			c.ofEntryOperations = mockOperations