Flexible pipeline

pkg/agent/controller/networkpolicy/networkpolicy_controller.go

@@ -143,7 +143,7 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 			c.ofClient.RegisterPacketInHandler(uint8(openflow.PacketInReasonNP), "dnsresponse", c.fqdnController)
 		}
 	}
-	c.reconciler = newReconciler(ofClient, ifaceStore, idAllocator, c.fqdnController, groupCounters, v4Enabled, v6Enabled)
+	c.reconciler = newReconciler(ofClient, ifaceStore, idAllocator, c.fqdnController, groupCounters, v4Enabled, v6Enabled, antreaPolicyEnabled)
 	c.ruleCache = newRuleCache(c.enqueueRule, podUpdateSubscriber, groupIDUpdates)
 	if statusManagerEnabled {
 		c.statusManager = newStatusController(antreaClientGetter, nodeName, c.ruleCache)

pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go

@@ -200,7 +200,20 @@ func newNetworkPolicyWithMultipleRules(name string, uid types.UID, from, to, app
 	}
 }
 
+func prepareMockTables() {
+	openflow.InitMockTables(
+		map[*openflow.Table]uint8{
+			openflow.AntreaPolicyEgressRuleTable:  uint8(5),
+			openflow.EgressRuleTable:              uint8(6),
+			openflow.EgressDefaultTable:           uint8(7),
+			openflow.AntreaPolicyIngressRuleTable: uint8(12),
+			openflow.IngressRuleTable:             uint8(13),
+			openflow.IngressDefaultTable:          uint8(14),
+		})
+}
+
 func TestAddSingleGroupRule(t *testing.T) {
+	prepareMockTables()
 	controller, clientset, reconciler := newTestController()
 	addressGroupWatcher := watch.NewFake()
 	appliedToGroupWatcher := watch.NewFake()
@@ -280,6 +293,7 @@ func TestAddSingleGroupRule(t *testing.T) {
 }
 
 func TestAddMultipleGroupsRule(t *testing.T) {
+	prepareMockTables()
 	controller, clientset, reconciler := newTestController()
 	addressGroupWatcher := watch.NewFake()
 	appliedToGroupWatcher := watch.NewFake()
@@ -359,6 +373,7 @@ func TestAddMultipleGroupsRule(t *testing.T) {
 }
 
 func TestDeleteRule(t *testing.T) {
+	prepareMockTables()
 	controller, clientset, reconciler := newTestController()
 	addressGroupWatcher := watch.NewFake()
 	appliedToGroupWatcher := watch.NewFake()
@@ -406,6 +421,7 @@ func TestDeleteRule(t *testing.T) {
 }
 
 func TestAddNetworkPolicyWithMultipleRules(t *testing.T) {
+	prepareMockTables()
 	controller, clientset, reconciler := newTestController()
 	addressGroupWatcher := watch.NewFake()
 	appliedToGroupWatcher := watch.NewFake()
@@ -488,6 +504,7 @@ func TestAddNetworkPolicyWithMultipleRules(t *testing.T) {
 }
 
 func TestNetworkPolicyMetrics(t *testing.T) {
+	prepareMockTables()
 	// Initialize NetworkPolicy metrics (prometheus)
 	metrics.InitializeNetworkPolicyMetrics()
 	controller, clientset, reconciler := newTestController()

pkg/agent/controller/networkpolicy/reconciler.go

@@ -219,16 +219,19 @@ func newReconciler(ofClient openflow.Client,
 	groupCounters []proxytypes.GroupCounter,
 	v4Enabled bool,
 	v6Enabled bool,
+	antreaPolicyEnabled bool,
 ) *reconciler {
 	priorityAssigners := map[uint8]*tablePriorityAssigner{}
-	for _, table := range openflow.GetAntreaPolicyBaselineTierTables() {
-		priorityAssigners[table.GetID()] = &tablePriorityAssigner{
-			assigner: newPriorityAssigner(true),
+	if antreaPolicyEnabled {
+		for _, table := range openflow.GetAntreaPolicyBaselineTierTables() {
+			priorityAssigners[table.GetID()] = &tablePriorityAssigner{
+				assigner: newPriorityAssigner(true),
+			}
 		}
-	}
-	for _, table := range openflow.GetAntreaPolicyMultiTierTables() {
-		priorityAssigners[table.GetID()] = &tablePriorityAssigner{
-			assigner: newPriorityAssigner(false),
+		for _, table := range openflow.GetAntreaPolicyMultiTierTables() {
+			priorityAssigners[table.GetID()] = &tablePriorityAssigner{
+				assigner: newPriorityAssigner(false),
+			}
 		}
 	}
 	reconciler := &reconciler{
@@ -297,7 +300,7 @@ func (r *reconciler) getOFRuleTable(rule *CompletedRule) uint8 {
 		}
 		return openflow.EgressRuleTable.GetID()
 	}
-	var ruleTables []binding.Table
+	var ruleTables []*openflow.Table
 	if rule.Direction == v1beta2.DirectionIn {
 		ruleTables = openflow.GetAntreaPolicyIngressTables()
 	} else {

pkg/agent/controller/networkpolicy/reconciler_test.go

@@ -103,11 +103,12 @@ func newTestReconciler(t *testing.T, controller *gomock.Controller, ifaceStore i
 	ch := make(chan string, 100)
 	groupIDAllocator := openflow.NewGroupAllocator(v6Enabled)
 	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(groupIDAllocator, ch)}
-	r := newReconciler(ofClient, ifaceStore, newIDAllocator(testAsyncDeleteInterval), f, groupCounters, v4Enabled, v6Enabled)
+	r := newReconciler(ofClient, ifaceStore, newIDAllocator(testAsyncDeleteInterval), f, groupCounters, v4Enabled, v6Enabled, true)
 	return r
 }
 
 func TestReconcilerForget(t *testing.T) {
+	prepareMockTables()
 	tests := []struct {
 		name              string
 		lastRealizeds     map[string]*lastRealized

pkg/agent/controller/traceflow/packetin.go

@@ -160,16 +160,17 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 		obs = append(obs, *ob)
 	}
 
-	// Collect Service DNAT and SNAT.
+	// Collect Service connections.
+	// - For packet is DNATed only, the final state is that ipDst != ctNwDst (in DNAT CT zone).
+	// - For packet is both DNATed and SNATed, the first state is also ipDst != ctNwDst (in DNAT CT zone), but the final
+	//   state is that ipSrc != ctNwSrc (in SNAT CT zone). The state in DNAT CT zone cannot be recognized in SNAT CT zone.
 	if !tfState.receiverOnly {
-		if isValidCtNw(ctNwDst) && ipDst != ctNwDst {
+		if isValidCtNw(ctNwDst) && ipDst != ctNwDst || isValidCtNw(ctNwSrc) && ipSrc != ctNwSrc {
 			ob := &crdv1alpha1.Observation{
 				Component:       crdv1alpha1.ComponentLB,
 				Action:          crdv1alpha1.ActionForwarded,
 				TranslatedDstIP: ipDst,
 			}
-			// Service SNAT can only happen alongside DNAT
-			// and only for hairpinned packets at the moment.
 			if isValidCtNw(ctNwSrc) && ipSrc != ctNwSrc {
 				ob.TranslatedSrcIP = ipSrc
 			}

pkg/agent/controller/traceflow/packetin_test.go

@@ -28,7 +28,24 @@ import (
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 )
 
+func prepareMockTables() {
+	openflow.InitMockTables(
+		map[*openflow.Table]uint8{
+			openflow.AntreaPolicyEgressRuleTable:  uint8(5),
+			openflow.EgressRuleTable:              uint8(6),
+			openflow.EgressDefaultTable:           uint8(7),
+			openflow.EgressMetricTable:            uint8(8),
+			openflow.AntreaPolicyIngressRuleTable: uint8(12),
+			openflow.IngressRuleTable:             uint8(13),
+			openflow.IngressDefaultTable:          uint8(14),
+			openflow.IngressMetricTable:           uint8(15),
+			openflow.L2ForwardingOutTable:         uint8(17),
+		})
+}
+
 func Test_getNetworkPolicyObservation(t *testing.T) {
+	prepareMockTables()
+
 	type args struct {
 		tableID uint8
 		ingress bool

pkg/agent/flowexporter/connections/conntrack_connections_test.go

@@ -36,7 +36,6 @@ import (
 	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
 	"antrea.io/antrea/pkg/agent/metrics"
 	"antrea.io/antrea/pkg/agent/openflow"
-	ofclient "antrea.io/antrea/pkg/agent/openflow"
 	proxytest "antrea.io/antrea/pkg/agent/proxy/testing"
 	agenttypes "antrea.io/antrea/pkg/agent/types"
 	cpv1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
@@ -83,7 +82,7 @@ var (
 		Priority:      nil,
 		Name:          "",
 		FlowID:        uint32(0),
-		TableID:       ofclient.IngressRuleTable.GetID(),
+		TableID:       uint8(10),
 		PolicyRef:     &np1,
 		EnableLogging: false,
 	}