Support AntreaNetworkPolicy reject action in Traceflow

pkg/agent/controller/traceflow/packetin.go

@@ -72,23 +72,34 @@ func (c *Controller) HandlePacketIn(pktIn *ofctrl.PacketIn) error {
 
 func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Traceflow, *crdv1alpha1.NodeResult, *crdv1alpha1.Packet, error) {
 	matchers := pktIn.GetMatches()
-	var match *ofctrl.MatchField
 
 	// Get data plane tag.
 	// Directly read data plane tag from packet.
+	var err error
 	var tag uint8
+	var ctNwDst, ipDst string
 	if pktIn.Data.Ethertype == protocol.IPv4_MSG {
 		ipPacket, ok := pktIn.Data.Data.(*protocol.IPv4)
 		if !ok {
 			return nil, nil, nil, errors.New("invalid traceflow IPv4 packet")
 		}
 		tag = ipPacket.DSCP
+		ctNwDst, err = getCTDstValue(matchers, false)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		ipDst = ipPacket.NWDst.String()
 	} else if pktIn.Data.Ethertype == protocol.IPv6_MSG {
 		ipv6Packet, ok := pktIn.Data.Data.(*protocol.IPv6)
 		if !ok {
 			return nil, nil, nil, errors.New("invalid traceflow IPv6 packet")
 		}
 		tag = ipv6Packet.TrafficClass >> 2
+		ctNwDst, err = getCTDstValue(matchers, true)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		ipDst = ipv6Packet.NWDst.String()
 	} else {
 		return nil, nil, nil, fmt.Errorf("unsupported traceflow packet Ethertype: %d", pktIn.Data.Ethertype)
 	}
@@ -139,32 +150,6 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	}
 
 	// Collect Service DNAT.
-	ctNwDst := ""
-	ipDst := ""
-	switch pktIn.Data.Ethertype {
-	case protocol.IPv4_MSG:
-		ipPacket, ok := pktIn.Data.Data.(*protocol.IPv4)
-		if !ok {
-			return nil, nil, nil, errors.New("invalid traceflow IPv4 packet")
-		}
-		ctNwDst, err = getCTDstValue(matchers, false)
-		if err != nil {
-			return nil, nil, nil, err
-		}
-		ipDst = ipPacket.NWDst.String()
-	case protocol.IPv6_MSG:
-		ipPacket, ok := pktIn.Data.Data.(*protocol.IPv6)
-		if !ok {
-			return nil, nil, nil, errors.New("invalid traceflow IPv6 packet")
-		}
-		ctNwDst, err = getCTDstValue(matchers, true)
-		if err != nil {
-			return nil, nil, nil, err
-		}
-		ipDst = ipPacket.NWDst.String()
-	default:
-		return nil, nil, nil, fmt.Errorf("unsupported traceflow packet ether type %d", pktIn.Data.Ethertype)
-	}
 	if isValidCtNw(ctNwDst) && ipDst != ctNwDst {
 		ob := &crdv1alpha1.Observation{
 			Component:       crdv1alpha1.ComponentLB,
@@ -175,7 +160,7 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	}
 
 	// Collect egress conjunctionID and get NetworkPolicy from cache.
-	if match = getMatchRegField(matchers, uint32(openflow.EgressReg)); match != nil {
+	if match := getMatchRegField(matchers, uint32(openflow.EgressReg)); match != nil {
 		egressInfo, err := getRegValue(match, nil)
 		if err != nil {
 			return nil, nil, nil, err
@@ -189,7 +174,7 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	}
 
 	// Collect ingress conjunctionID and get NetworkPolicy from cache.
-	if match = getMatchRegField(matchers, uint32(openflow.IngressReg)); match != nil {
+	if match := getMatchRegField(matchers, uint32(openflow.IngressReg)); match != nil {
 		ingressInfo, err := getRegValue(match, nil)
 		if err != nil {
 			return nil, nil, nil, err
@@ -205,14 +190,18 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	// Get drop table.
 	if tableID == uint8(openflow.EgressMetricTable) || tableID == uint8(openflow.IngressMetricTable) {
 		ob := getNetworkPolicyObservation(tableID, tableID == uint8(openflow.IngressMetricTable))
-		if match = getMatchRegField(matchers, uint32(openflow.CNPDenyConjIDReg)); match != nil {
+		if match := getMatchRegField(matchers, uint32(openflow.CNPDenyConjIDReg)); match != nil {
 			notAllowConjInfo, err := getRegValue(match, nil)
 			if err != nil {
 				return nil, nil, nil, err
 			}
-			npRef := c.networkPolicyQuerier.GetNetworkPolicyByRuleFlowID(notAllowConjInfo)
-			if npRef != nil {
-				ob.NetworkPolicy = npRef.ToString()
+			if ruleRef := c.networkPolicyQuerier.GetRuleByFlowID(notAllowConjInfo); ruleRef != nil {
+				if npRef := ruleRef.PolicyRef; npRef != nil {
+					ob.NetworkPolicy = npRef.ToString()
+				}
+				if ruleRef.Action != nil && *ruleRef.Action == crdv1alpha1.RuleActionReject {
+					ob.Action = crdv1alpha1.ActionRejected
+				}
 			}
 		}
 		obs = append(obs, *ob)
@@ -226,14 +215,14 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 		ob := new(crdv1alpha1.Observation)
 		tunnelDstIP := ""
 		isIPv6 := c.nodeConfig.NodeIPAddr.IP.To4() == nil
-		if match = getMatchTunnelDstField(matchers, isIPv6); match != nil {
+		if match := getMatchTunnelDstField(matchers, isIPv6); match != nil {
 			tunnelDstIP, err = getTunnelDstValue(match)
 			if err != nil {
 				return nil, nil, nil, err
 			}
 		}
 		var outputPort uint32
-		if match = getMatchRegField(matchers, uint32(openflow.PortCacheReg)); match != nil {
+		if match := getMatchRegField(matchers, uint32(openflow.PortCacheReg)); match != nil {
 			outputPort, err = getRegValue(match, nil)
 			if err != nil {
 				return nil, nil, nil, err

pkg/apis/crd/v1alpha1/types.go

@@ -46,6 +46,7 @@ const (
 	ActionReceived  TraceflowAction = "Received"
 	ActionForwarded TraceflowAction = "Forwarded"
 	ActionDropped   TraceflowAction = "Dropped"
+	ActionRejected  TraceflowAction = "Rejected"
 	// ActionForwardedOutOfOverlay indicates that the packet has been forwarded out of the network
 	// managed by Antrea. This indicates that the Traceflow request can be considered complete.
 	ActionForwardedOutOfOverlay TraceflowAction = "ForwardedOutOfOverlay"

pkg/controller/traceflow/controller.go

@@ -299,6 +299,7 @@ func (c *Controller) checkTraceflowStatus(tf *crdv1alpha1.Traceflow) error {
 				}
 				if ob.Action == crdv1alpha1.ActionDelivered ||
 					ob.Action == crdv1alpha1.ActionDropped ||
+					ob.Action == crdv1alpha1.ActionRejected ||
 					ob.Action == crdv1alpha1.ActionForwardedOutOfOverlay {
 					receiver = true
 				}

test/e2e/traceflow_test.go

@@ -80,19 +80,29 @@ func TestTraceflowIntraNodeANP(t *testing.T) {
 	failOnError(err, t)
 
 	node1 := nodeName(0)
-	node1Pods, _, node1CleanupFn := createTestBusyboxPods(t, data, 2, node1)
+	node1Pods, _, node1CleanupFn := createTestBusyboxPods(t, data, 3, node1)
 	defer node1CleanupFn()
 
 	var denyIngress *secv1alpha1.NetworkPolicy
 	denyIngressName := "test-anp-deny-ingress"
-	if denyIngress, err = data.createANPDenyIngress("antrea-e2e", node1Pods[1], denyIngressName); err != nil {
+	if denyIngress, err = data.createANPDenyIngress("antrea-e2e", node1Pods[1], denyIngressName, false); err != nil {
 		t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
 	}
 	defer func() {
 		if err = data.deleteAntreaNetworkpolicy(denyIngress); err != nil {
 			t.Errorf("Error when deleting Antrea NetworkPolicy: %v", err)
 		}
 	}()
+	var rejectIngress *secv1alpha1.NetworkPolicy
+	rejectIngressName := "test-anp-reject-ingress"
+	if rejectIngress, err = data.createANPDenyIngress("antrea-e2e", node1Pods[2], rejectIngressName, true); err != nil {
+		t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
+	}
+	defer func() {
+		if err = data.deleteAntreaNetworkpolicy(rejectIngress); err != nil {
+			t.Errorf("Error when deleting Antrea NetworkPolicy: %v", err)
+		}
+	}()
 	antreaPod, err := data.getAntreaPodOnNode(node1)
 	if err = data.waitForNetworkpolicyRealized(antreaPod, denyIngressName, v1beta2.AntreaNetworkPolicy); err != nil {
 		t.Fatal(err)
@@ -146,6 +156,53 @@ func TestTraceflowIntraNodeANP(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:      "ANPRejectIngressIPv4",
+			ipVersion: 4,
+			tf: &v1alpha1.Traceflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: randName(fmt.Sprintf("%s-%s-to-%s-%s-", testNamespace, node1Pods[0], testNamespace, node1Pods[2])),
+				},
+				Spec: v1alpha1.TraceflowSpec{
+					Source: v1alpha1.Source{
+						Namespace: testNamespace,
+						Pod:       node1Pods[0],
+					},
+					Destination: v1alpha1.Destination{
+						Namespace: testNamespace,
+						Pod:       node1Pods[2],
+					},
+					Packet: v1alpha1.Packet{
+						IPHeader: v1alpha1.IPHeader{
+							Protocol: protocolTCP,
+						},
+						TransportHeader: v1alpha1.TransportHeader{
+							TCP: &v1alpha1.TCPHeader{
+								DstPort: 80,
+								Flags:   2,
+							},
+						},
+					},
+				},
+			},
+			expectedPhase: v1alpha1.Succeeded,
+			expectedResults: []v1alpha1.NodeResult{
+				{
+					Node: node1,
+					Observations: []v1alpha1.Observation{
+						{
+							Component: v1alpha1.ComponentSpoofGuard,
+							Action:    v1alpha1.ActionForwarded,
+						},
+						{
+							Component:     v1alpha1.ComponentNetworkPolicy,
+							ComponentInfo: "IngressMetric",
+							Action:        v1alpha1.ActionRejected,
+						},
+					},
+				},
+			},
+		},
 		{
 			name:      "ANPDenyIngressIPv6",
 			ipVersion: 6,
@@ -1759,8 +1816,11 @@ func compareObservations(expected v1alpha1.NodeResult, actual v1alpha1.NodeResul
 }
 
 // createANPDenyIngress creates an Antrea NetworkPolicy that denies ingress traffic for pods of specific label.
-func (data *TestData) createANPDenyIngress(key string, value string, name string) (*secv1alpha1.NetworkPolicy, error) {
+func (data *TestData) createANPDenyIngress(key string, value string, name string, isReject bool) (*secv1alpha1.NetworkPolicy, error) {
 	dropACT := secv1alpha1.RuleActionDrop
+	if isReject {
+		dropACT = secv1alpha1.RuleActionReject
+	}
 	anp := secv1alpha1.NetworkPolicy{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: name,