Repository: antrea-io/antrea
Pull request #4206 (2022-09-06 11:15:15)

[multicast]fix IGMP v1 report matches flow in MulticastEgressRule
This is because MulticastEgressRule defines rule to match nw_src and nw_dst, while
IGMP v1 report will sent to multicast group address. For example, if we define a
rule like:

```
spec:
    priority: 5
    tier: securityops
    appliedTo:
      - podSelector:
          matchLabels:
            app: mcjoin6
    egress:
      - action: Drop
        to:
          - ipBlock:
              cidr: 225.1.2.4/32

```
Not only normal multicast traffic from Pods match label "app=mcjoin6", but also IGMP
v1 report will be matched. In this case, if there is egress rule to block multicast
traffic, IGMP v1 report will also be blocked, which is not expected.
Another issue, is that multicast stats will also count IGMP v1 report packet into
normal multicast egress rule as well. This

This patch defines a higher priority rule to match IGMP report to jump to table
MulticastRouting directly.

Signed-off-by: Bin Liu <biliu@vmware.com>

liu4480: /test-conformance
/test-e2e
/test-multicluster-e2e
/test-networkpolicy

liu4480: /test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
XinShuYang: /test-conformance
 /test-e2e
 /test-networkpolicy
 /test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
ceclinux: LGTM
liu4480: /test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
liu4480: /test-ipv6-conformance
/test-ipv6-e2e
/test-ipv6-only-e2e
liu4480: /test-multicluster-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
liu4480: /test-conformance
/test-e2e
/test-networkpolicy
/test-multicast-e2e
liu4480: /test-e2e 
tnqn: The commit message is a little confusing:
> Not only normal multicast traffic from Pods match label "app=mcjoin6", but also IGMP
v1 report will be matched. In this case, if there is egress rule to block multicast
traffic, IGMP v1 report will also be blocked, which is not expected.

What example it wants to show? allow or block? The yaml is about allow but the words are about block.

There is one extra "this" in the commit message.

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 311f761a7..1a4d7c681 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -13,13 +13,13 @@ linters-settings:
     ignore-generated-header: false
     severity: warning
     confidence: 0.8
-    # errorf, var-declaration temporarily disabled because of https://github.com/golangci/golangci-lint/issues/2997
     rules:
       - name: unreachable-code
-      # - name: errorf
+      - name: errorf
       - name: range
       - name: superfluous-else
-      # - name: var-declaration
+      - name: var-declaration
+      - name: duplicated-imports
 
 linters:
   disable-all: true
diff --git a/Makefile b/Makefile
index 62bfb1a48..e2b3e21ef 100644
--- a/Makefile
+++ b/Makefile
@@ -261,7 +261,7 @@ fmt:
 
 .golangci-bin:
 	@echo "===> Installing Golangci-lint <==="
-	@curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $@ v1.48.0
+	@curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $@ v1.50.0
 
 .PHONY: golangci
 golangci: .golangci-bin
diff --git a/docs/antrea-network-policy.md b/docs/antrea-network-policy.md
index e6d2f89f6..5fbe7e3d7 100644
--- a/docs/antrea-network-policy.md
+++ b/docs/antrea-network-policy.md
@@ -636,7 +636,7 @@ to select Pods. More details can be found in the [ServiceAccountSelector](#servi
 **Note**: The order in which the egress rules are specified matters, i.e., rules will
 be enforced in the order in which they are written.
 
-**enableLogging**: A ClusterNetworkPolicy ingress or egress rule can be
+**enableLogging**: Antrea-native policy ingress or egress rules can be
 audited by enabling its logging field. When `enableLogging` field is set to
 true, the first packet of any connection that matches this rule will be logged
 to a separate file (`/var/log/antrea/networkpolicy/np.log`) on the Node on
@@ -645,16 +645,19 @@ analysis. By default, rules are not logged. The example policy logs all
 traffic that matches the "DropToThirdParty" egress rule, while the rule
 "AllowFromFrontend" is not logged. Specifically for drop and reject rules,
 deduplication is applied to reduce duplicated logs, and duplication buffer
-length is set to 1 second. The rules are logged in the following format:
+length is set to 1 second. If a rule name is not provided, an identifiable
+name will be generated for the rule and displayed in the log.
+The rules are logged in the following format:
 
 ```text
-    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length>
+    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <rule-name> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length>
     Deduplication:
-    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length> [<num of packets> packets in <duplicate duration>]
+    <yyyy/mm/dd> <time> <ovs-table-name> <antrea-native-policy-reference> <rule-name> <action> <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length> [<num of packets> packets in <duplicate duration>]
 
     Examples:
-    2020/11/02 22:21:21.148395 AntreaPolicyAppTierIngressRule AntreaNetworkPolicy:default/test-anp Allow 61800 10.10.1.65 35402 10.0.0.5 80 TCP 60
-    2021/06/24 23:56:41.346165 AntreaPolicyEgressRule AntreaNetworkPolicy:default/test-anp Drop 44900 10.10.1.65 35402 10.0.0.5 80 TCP 60 [3 packets in 1.011379442s]
+    2020/11/02 22:21:21.148395 AntreaPolicyAppTierIngressRule AntreaNetworkPolicy:default/test-anp test-rule Allow 61800 10.10.1.65 35402 10.0.0.5 80 TCP 60
+    2021/06/24 23:56:41.346165 AntreaPolicyEgressRule AntreaNetworkPolicy:default/test-anp test-rule Drop 44900 10.10.1.65 35402 10.0.0.5 80 TCP 60 [3 packets in 1.011379442s]
+    2022/09/20 02:21:25.879364 AntreaPolicyIngressRule AntreaNetworkPolicy:default/test-anp ingress-drop-1cffec1 Drop 44900 10.10.1.14 <nil> 10.10.1.15 <nil> ICMP 84
 ```
 
 Kubernetes NetworkPolicies can also be audited using Antrea logging to the same file
@@ -663,17 +666,18 @@ Kubernetes NetworkPolicies can also be audited using Antrea logging to the same
 for all NetworkPolicies in the Namespace. Packets of any connection that match
 a NetworkPolicy rule will be logged with a reference to the NetworkPolicy name,
 but packets dropped by the implicit "default drop" (not allowed by any NetworkPolicy)
-will only be logged with consistent name `K8sNetworkPolicy` for reference.
-The rules are logged in the following format:
+will only be logged with consistent name `K8sNetworkPolicy` for reference. When
+using Antrea logging for Kubernetes NetworkPolicies, the rule name field is not
+set and defaults to `<nil>` value. The rules are logged in the following format:
 
 ```text
-    <yyyy/mm/dd> <time> <ovs-table-name> <k8s-network-policy-reference> Allow <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length>
+    <yyyy/mm/dd> <time> <ovs-table-name> <k8s-network-policy-reference> <nil> Allow <openflow-priority> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length>
     Default dropped traffic:
-    <yyyy/mm/dd> <time> <ovs-table-name> K8sNetworkPolicy Drop -1 <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length> [<num of packets> packets in <duplicate duration>]
+    <yyyy/mm/dd> <time> <ovs-table-name> K8sNetworkPolicy <nil> Drop <nil> <source-ip> <source-port> <destination-ip> <destination-port> <protocol> <packet-length> [<num of packets> packets in <duplicate duration>]
 
     Examples:
-    2022/07/26 06:55:56.170456 IngressRule K8sNetworkPolicy:default/test-np-log Allow 190 10.10.1.82 49518 10.10.1.84 80 TCP 60
-    2022/07/26 06:55:57.142206 IngressDefaultRule K8sNetworkPolicy Drop -1 10.10.1.83 38608 10.10.1.84 80 TCP 60
+    2022/07/26 06:55:56.170456 IngressRule K8sNetworkPolicy:default/test-np-log <nil> Allow 190 10.10.1.82 49518 10.10.1.84 80 TCP 60
+    2022/07/26 06:55:57.142206 IngressDefaultRule K8sNetworkPolicy <nil> Drop <nil> 10.10.1.83 38608 10.10.1.84 80 TCP 60
 ```
 
 Fluentd can be used to assist with collecting and analyzing the logs. Refer to the
diff --git a/go.mod b/go.mod
index feee895a6..dad076898 100644
--- a/go.mod
+++ b/go.mod
@@ -42,7 +42,7 @@ require (
 	github.com/miekg/dns v1.1.50
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 	github.com/onsi/ginkgo v1.16.5
-	github.com/onsi/gomega v1.20.2
+	github.com/onsi/gomega v1.21.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.13.0
 	github.com/prometheus/common v0.37.0
@@ -63,7 +63,7 @@ require (
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8
 	golang.org/x/tools v0.1.10-0.20220218145154-897bd77cd717
 	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20210506160403-92e472f520a5
-	google.golang.org/grpc v1.49.0
+	google.golang.org/grpc v1.50.0
 	google.golang.org/protobuf v1.28.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v2 v2.4.0
diff --git a/go.sum b/go.sum
index fcf7b06e5..664e44d64 100644
--- a/go.sum
+++ b/go.sum
@@ -748,8 +748,8 @@ github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
-github.com/onsi/gomega v1.20.2 h1:8uQq0zMgLEfa0vRrrBgaJF2gyW9Da9BmfGV+OyUzfkY=
-github.com/onsi/gomega v1.20.2/go.mod h1:iYAIXgPSaDHak0LCMA+AWBpIKBr8WZicMxnE8luStNc=
+github.com/onsi/gomega v1.21.1 h1:OB/euWYIExnPBohllTicTHmGTrMaqJ67nIu80j0/uEM=
+github.com/onsi/gomega v1.21.1/go.mod h1:iYAIXgPSaDHak0LCMA+AWBpIKBr8WZicMxnE8luStNc=
 github.com/opencontainers/go-digest v0.0.0-20180430190053-c9281466c8b2/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/runc v0.0.0-20190115041553-12f6a991201f/go.mod h1:qT5XzbpPznkRYVz/mWwUaVBUv2rmF59PVA73FjuZG0U=
@@ -1486,8 +1486,8 @@ google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAG
 google.golang.org/grpc v1.37.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
-google.golang.org/grpc v1.49.0 h1:WTLtQzmQori5FUH25Pq4WT22oCsv8USpQ+F6rqtsmxw=
-google.golang.org/grpc v1.49.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
+google.golang.org/grpc v1.50.0 h1:fPVVDxY9w++VjTZsYvXWqEf9Rqar/e+9zYfxKK+W+YU=
+google.golang.org/grpc v1.50.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
diff --git a/pkg/agent/agent_windows.go b/pkg/agent/agent_windows.go
index 615cf41d1..054817f00 100644
--- a/pkg/agent/agent_windows.go
+++ b/pkg/agent/agent_windows.go
@@ -31,7 +31,6 @@ import (
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/ovs/ovsctl"
-	"antrea.io/antrea/pkg/util/ip"
 	utilip "antrea.io/antrea/pkg/util/ip"
 )
 
@@ -65,7 +64,7 @@ func (i *Initializer) prepareHNSNetworkAndOVSExtension() error {
 	// Get uplink network configuration. The uplink interface is the one used for transporting Pod traffic across Nodes.
 	// Use the interface specified with "transportInterface" in the configuration if configured, otherwise the interface
 	// configured with NodeIP is used as uplink.
-	_, _, adapter, err := i.getNodeInterfaceFromIP(&ip.DualStackIPs{IPv4: i.nodeConfig.NodeTransportIPv4Addr.IP})
+	_, _, adapter, err := i.getNodeInterfaceFromIP(&utilip.DualStackIPs{IPv4: i.nodeConfig.NodeTransportIPv4Addr.IP})
 	if err != nil {
 		return err
 	}
diff --git a/pkg/agent/controller/networkpolicy/audit_logging.go b/pkg/agent/controller/networkpolicy/audit_logging.go
index c29190374..51424a913 100644
--- a/pkg/agent/controller/networkpolicy/audit_logging.go
+++ b/pkg/agent/controller/networkpolicy/audit_logging.go
@@ -66,7 +66,8 @@ type AntreaPolicyLogger struct {
 // logInfo will be set by retrieving info from packetin and register.
 type logInfo struct {
 	tableName   string // name of the table sending packetin
-	npRef       string // Network Policy name reference for Antrea NetworkPolicy
+	npRef       string // Network Policy name reference
+	ruleName    string // Network Policy rule name for Antrea-native policies
 	disposition string // Allow/Drop of the rule sending packetin
 	ofPriority  string // openflow priority of the flow sending packetin
 	srcIP       string // source IP of the traffic logged
@@ -134,7 +135,7 @@ func (l *AntreaPolicyLogger) updateLogKey(logMsg string, bufferLength time.Durat
 // LogDedupPacket logs information in ob based on disposition and duplication conditions.
 func (l *AntreaPolicyLogger) LogDedupPacket(ob *logInfo) {
 	// Deduplicate non-Allow packet log.
-	logMsg := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %d", ob.tableName, ob.npRef, ob.disposition, ob.ofPriority, ob.srcIP, ob.srcPort, ob.destIP, ob.destPort, ob.protocolStr, ob.pktLength)
+	logMsg := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %s %d", ob.tableName, ob.npRef, ob.ruleName, ob.disposition, ob.ofPriority, ob.srcIP, ob.srcPort, ob.destIP, ob.destPort, ob.protocolStr, ob.pktLength)
 	if ob.disposition == openflow.DispositionToString[openflow.DispositionAllow] {
 		l.anpLogger.Printf(logMsg)
 	} else {
@@ -188,29 +189,51 @@ func getNetworkPolicyInfo(pktIn *ofctrl.PacketIn, c *Controller, ob *logInfo) er
 
 	// Get disposition Allow or Drop.
 	match = getMatchRegField(matchers, openflow.APDispositionField)
-	info, err := getInfoInReg(match, openflow.APDispositionField.GetRange().ToNXRange())
+	disposition, err := getInfoInReg(match, openflow.APDispositionField.GetRange().ToNXRange())
 	if err != nil {
 		return fmt.Errorf("received error while unloading disposition from reg: %v", err)
 	}
-	ob.disposition = openflow.DispositionToString[info]
+	ob.disposition = openflow.DispositionToString[disposition]
 
 	// Set match to corresponding ingress/egress reg according to disposition.
-	match = getMatch(matchers, tableID, info)
+	match = getMatch(matchers, tableID, disposition)
 
 	if match != nil {
 		// Get NetworkPolicy full name and OF priority of the conjunction.
-		info, err = getInfoInReg(match, nil)
+		conjID, err := getInfoInReg(match, nil)
 		if err != nil {
 			return fmt.Errorf("received error while unloading conjunction id from reg: %v", err)
 		}
-		ob.npRef, ob.ofPriority = c.ofClient.GetPolicyInfoFromConjunction(info)
+		ob.npRef, ob.ofPriority, ob.ruleName = c.ofClient.GetPolicyInfoFromConjunction(conjID)
+		if ob.npRef == "" || ob.ofPriority == "" {
+			return fmt.Errorf("networkpolicy not found for conjunction id: %v", conjID)
+		}
+		// Placeholder for K8s NetworkPolicies without rule names.
+		if ob.ruleName == "" {
+			ob.ruleName = "<nil>"
+		}
 	} else {
 		// For K8s NetworkPolicy implicit drop action, we cannot get Namespace/name.
-		ob.npRef, ob.ofPriority = string(v1beta2.K8sNetworkPolicy), "-1"
+		ob.npRef, ob.ofPriority, ob.ruleName = string(v1beta2.K8sNetworkPolicy), "<nil>", "<nil>"
 	}
 	return nil
 }
 
+// getPacketInfo fills in IP, packet length, protocol, port number of logInfo ob.
+func getPacketInfo(packet *binding.Packet, ob *logInfo) {
+	ob.srcIP = packet.SourceIP.String()
+	ob.destIP = packet.DestinationIP.String()
+	ob.pktLength = packet.IPLength
+	ob.protocolStr = ip.IPProtocolNumberToString(packet.IPProto, "UnknownProtocol")
+	if ob.protocolStr == "TCP" || ob.protocolStr == "UDP" {
+		ob.srcPort = strconv.Itoa(int(packet.SourcePort))
+		ob.destPort = strconv.Itoa(int(packet.DestinationPort))
+	} else {
+		// Placeholders for ICMP packets without port numbers.
+		ob.srcPort, ob.destPort = "<nil>", "<nil>"
+	}
+}
+
 // logPacket retrieves information from openflow reg, controller cache, packet-in
 // packet to log. Log is deduplicated for non-Allow packets from record in logDeduplication.
 // Deduplication is safe guarded by logRecordDedupMap mutex.
@@ -226,17 +249,7 @@ func (c *Controller) logPacket(pktIn *ofctrl.PacketIn) error {
 	if err != nil {
 		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
 	}
-	ob.srcIP = packet.SourceIP.String()
-	ob.destIP = packet.DestinationIP.String()
-	ob.pktLength = packet.IPLength
-	ob.protocolStr = ip.IPProtocolNumberToString(packet.IPProto, "UnknownProtocol")
-	if ob.protocolStr == "TCP" || ob.protocolStr == "UDP" {
-		ob.srcPort = strconv.Itoa(int(packet.SourcePort))
-		ob.destPort = strconv.Itoa(int(packet.DestinationPort))
-	} else {
-		// Placeholders for ICMP packets without port numbers.
-		ob.srcPort, ob.destPort = "<nil>", "<nil>"
-	}
+	getPacketInfo(packet, ob)
 
 	// Log the ob info to corresponding file w/ deduplication.
 	c.antreaPolicyLogger.LogDedupPacket(ob)
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index 034f7471b..1a61ca5ca 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -211,7 +211,7 @@ type Client interface {
 	InitialTLVMap() error
 
 	// Find Network Policy reference and OFpriority by conjunction ID.
-	GetPolicyInfoFromConjunction(ruleID uint32) (string, string)
+	GetPolicyInfoFromConjunction(ruleID uint32) (string, string, string)
 
 	// RegisterPacketInHandler uses SubscribePacketIn to get PacketIn message and process received
 	// packets through registered handlers.
@@ -1220,6 +1220,9 @@ func (c *client) InstallMulticastInitialFlows(pktInReason uint8) error {
 	flows := c.featureMulticast.igmpPktInFlows(pktInReason)
 	flows = append(flows, c.featureMulticast.externalMulticastReceiverFlow())
 	flows = append(flows, c.featureMulticast.multicastSkipIGMPMetricFlows()...)
+	if c.enableAntreaPolicy {
+		flows = append(flows, c.featureMulticast.igmpEgressFlow())
+	}
 	cacheKey := "multicast"
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
diff --git a/pkg/agent/openflow/framework.go b/pkg/agent/openflow/framework.go
index f4a24d945..76d006594 100644
--- a/pkg/agent/openflow/framework.go
+++ b/pkg/agent/openflow/framework.go
@@ -216,12 +216,10 @@ func (f *featureNetworkPolicy) getRequiredTables() []*Table {
 		if f.enableMulticast {
 			tables = append(tables,
 				MulticastEgressRuleTable,
+				MulticastEgressPodMetricTable,
 				MulticastEgressMetricTable,
 				MulticastIngressRuleTable,
 				MulticastIngressPodMetricTable,
-				MulticastEgressRuleTable,
-				MulticastEgressMetricTable,
-				MulticastEgressPodMetricTable,
 				MulticastIngressMetricTable,
 			)
 		}
diff --git a/pkg/agent/openflow/network_policy.go b/pkg/agent/openflow/network_policy.go
index 0d6d0f819..784bddbcd 100644
--- a/pkg/agent/openflow/network_policy.go
+++ b/pkg/agent/openflow/network_policy.go
@@ -639,6 +639,7 @@ type policyRuleConjunction struct {
 	// NetworkPolicy reference information for debugging usage, its value can be nil
 	// for conjunctions that are not built for a specific NetworkPolicy, e.g. DNS packetin Conjunction.
 	npRef       *v1beta2.NetworkPolicyReference
+	ruleName    string
 	ruleTableID uint8
 }
 
@@ -1109,8 +1110,9 @@ func (f *featureNetworkPolicy) calculateActionFlowChangesForRule(rule *types.Pol
 		return nil
 	}
 	conj = &policyRuleConjunction{
-		id:    ruleOfID,
-		npRef: rule.PolicyRef,
+		id:       ruleOfID,
+		npRef:    rule.PolicyRef,
+		ruleName: rule.Name,
 	}
 	nClause, ruleTable, dropTable := conj.calculateClauses(rule)
 	conj.ruleTableID = rule.TableID
@@ -1450,16 +1452,16 @@ func (f *featureNetworkPolicy) getPolicyRuleConjunction(ruleID uint32) *policyRu
 	return conj.(*policyRuleConjunction)
 }
 
-func (c *client) GetPolicyInfoFromConjunction(ruleID uint32) (string, string) {
+func (c *client) GetPolicyInfoFromConjunction(ruleID uint32) (string, string, string) {
 	conjunction := c.featureNetworkPolicy.getPolicyRuleConjunction(ruleID)
 	if conjunction == nil {
-		return "", ""
+		return "", "", ""
 	}
 	priorities := conjunction.ActionFlowPriorities()
 	if len(priorities) == 0 {
-		return "", ""
+		return "", "", ""
 	}
-	return conjunction.npRef.ToString(), priorities[0]
+	return conjunction.npRef.ToString(), priorities[0], conjunction.ruleName
 }
 
 // UninstallPolicyRuleFlows removes the Openflow entry relevant to the specified NetworkPolicy rule.
@@ -1709,6 +1711,7 @@ func (f *featureNetworkPolicy) updateConjunctionActionFlows(conj *policyRuleConj
 		serviceClause: conj.serviceClause,
 		actionFlows:   newActionFlows,
 		npRef:         conj.npRef,
+		ruleName:      conj.ruleName,
 		ruleTableID:   conj.ruleTableID,
 	}
 	return newConj
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index 5fa6317d8..6239b254d 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -2579,6 +2579,18 @@ func pipelineClassifyFlow(cookieID uint64, protocol binding.Protocol, pipeline b
 		Done()
 }
 
+// igmpEgressFlow generates flows to match IGMP report to jump to table MulticastRoutingTable.
+// This is because normal multicast egress rule can match IGMP v1 report, when there is egress
+// rule to block multicast traffic, IGMP v1 report will also be blocked, which is not expected.
+func (f *featureMulticast) igmpEgressFlow() binding.Flow {
+	return MulticastEgressRuleTable.ofTable.BuildFlow(priorityTopAntreaPolicy).
+		Cookie(f.cookieAllocator.Request(f.category).Raw()).
+		MatchProtocol(binding.ProtocolIGMP).
+		MatchRegMark(FromLocalRegMark).
+		Action().GotoStage(stageRouting).
+		Done()
+}
+
 // igmpPktInFlows generates the flow to load CustomReasonIGMPRegMark to mark the IGMP packet in MulticastRoutingTable
 // and sends it to antrea-agent.
 func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
diff --git a/pkg/agent/proxy/proxier.go b/pkg/agent/proxy/proxier.go
index 036124794..8394e24fc 100644
--- a/pkg/agent/proxy/proxier.go
+++ b/pkg/agent/proxy/proxier.go
@@ -27,7 +27,6 @@ import (
 	discovery "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	apimachinerytypes "k8s.io/apimachinery/pkg/types"
-	k8sapitypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
@@ -863,7 +862,7 @@ func (p *proxier) GetProxyProvider() k8sproxy.Provider {
 }
 
 func (p *proxier) GetServiceFlowKeys(serviceName, namespace string) ([]string, []binding.GroupIDType, bool) {
-	namespacedName := k8sapitypes.NamespacedName{Namespace: namespace, Name: serviceName}
+	namespacedName := apimachinerytypes.NamespacedName{Namespace: namespace, Name: serviceName}
 	p.serviceEndpointsMapsMutex.Lock()
 	defer p.serviceEndpointsMapsMutex.Unlock()
 
diff --git a/pkg/controller/certificatesigningrequest/ipsec_csr_signing_controller.go b/pkg/controller/certificatesigningrequest/ipsec_csr_signing_controller.go
index e6fc51bc2..2f2e847d7 100644
--- a/pkg/controller/certificatesigningrequest/ipsec_csr_signing_controller.go
+++ b/pkg/controller/certificatesigningrequest/ipsec_csr_signing_controller.go
@@ -38,7 +38,6 @@ import (
 	corev1informers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	csrlister "k8s.io/client-go/listers/certificates/v1"
-	csrlisters "k8s.io/client-go/listers/certificates/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	certutil "k8s.io/client-go/util/cert"
@@ -110,7 +109,7 @@ func (c *certificateAuthority) signCSR(template *x509.Certificate, requestKey cr
 }
 
 // NewIPsecCSRSigningController returns a new *IPsecCSRSigningController.
-func NewIPsecCSRSigningController(client clientset.Interface, csrInformer cache.SharedIndexInformer, csrLister csrlisters.CertificateSigningRequestLister, selfSignedCA bool) *IPsecCSRSigningController {
+func NewIPsecCSRSigningController(client clientset.Interface, csrInformer cache.SharedIndexInformer, csrLister csrlister.CertificateSigningRequestLister, selfSignedCA bool) *IPsecCSRSigningController {
 
 	caConfigMapInformer := corev1informers.NewFilteredConfigMapInformer(client, env.GetAntreaNamespace(), resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, func(listOptions *metav1.ListOptions) {
 		listOptions.FieldSelector = fields.OneTermEqualSelector("metadata.name", ipsecRootCAName).String()
diff --git a/pkg/controller/serviceexternalip/controller.go b/pkg/controller/serviceexternalip/controller.go
index 12ebb858e..e8276bffa 100644
--- a/pkg/controller/serviceexternalip/controller.go
+++ b/pkg/controller/serviceexternalip/controller.go
@@ -24,7 +24,6 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
 	apimachineryerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -182,7 +181,7 @@ func (c *ServiceExternalIPController) restoreIPAllocations(services []*corev1.Se
 		}
 		ip := net.ParseIP(svc.Status.LoadBalancer.Ingress[0].IP)
 		allocation := externalippool.IPAllocation{
-			ObjectReference: v1.ObjectReference{
+			ObjectReference: corev1.ObjectReference{
 				Name:      svc.Name,
 				Namespace: svc.Namespace,
 				Kind:      svc.Kind,
diff --git a/plugins/octant/go.mod b/plugins/octant/go.mod
index 5dfe3c3bc..00a193fd9 100644
--- a/plugins/octant/go.mod
+++ b/plugins/octant/go.mod
@@ -112,7 +112,7 @@ require (
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20220107163113-42d7afdf6368 // indirect
-	google.golang.org/grpc v1.49.0 // indirect
+	google.golang.org/grpc v1.50.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
diff --git a/plugins/octant/go.sum b/plugins/octant/go.sum
index 106c186a6..ab8159aea 100644
--- a/plugins/octant/go.sum
+++ b/plugins/octant/go.sum
@@ -666,7 +666,7 @@ github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
-github.com/onsi/gomega v1.20.2 h1:8uQq0zMgLEfa0vRrrBgaJF2gyW9Da9BmfGV+OyUzfkY=
+github.com/onsi/gomega v1.21.1 h1:OB/euWYIExnPBohllTicTHmGTrMaqJ67nIu80j0/uEM=
 github.com/opencontainers/go-digest v0.0.0-20170106003457-a6d0ee40d420/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
 github.com/opencontainers/go-digest v0.0.0-20180430190053-c9281466c8b2/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
 github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
@@ -1342,8 +1342,8 @@ google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAG
 google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
-google.golang.org/grpc v1.49.0 h1:WTLtQzmQori5FUH25Pq4WT22oCsv8USpQ+F6rqtsmxw=
-google.golang.org/grpc v1.49.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
+google.golang.org/grpc v1.50.0 h1:fPVVDxY9w++VjTZsYvXWqEf9Rqar/e+9zYfxKK+W+YU=
+google.golang.org/grpc v1.50.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=


Test patch:
diff --git a/pkg/agent/controller/networkpolicy/audit_logging_test.go b/pkg/agent/controller/networkpolicy/audit_logging_test.go
index 36323b616..5388a7259 100644
--- a/pkg/agent/controller/networkpolicy/audit_logging_test.go
+++ b/pkg/agent/controller/networkpolicy/audit_logging_test.go
@@ -18,14 +18,23 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"net"
 	"strconv"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"antrea.io/libOpenflow/openflow15"
+	"antrea.io/ofnet/ofctrl"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"antrea.io/antrea/pkg/agent/openflow"
+	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
+	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+	"antrea.io/antrea/pkg/util/ip"
 )
 
 const (
@@ -115,6 +124,7 @@ func newLogInfo(disposition string) (*logInfo, string) {
 	testLogInfo := &logInfo{
 		tableName:   "AntreaPolicyIngressRule",
 		npRef:       "AntreaNetworkPolicy:default/test",
+		ruleName:    "test-rule",
 		ofPriority:  "0",
 		disposition: disposition,
 		srcIP:       "0.0.0.0",
@@ -124,8 +134,8 @@ func newLogInfo(disposition string) (*logInfo, string) {
 		protocolStr: "TCP",
 		pktLength:   60,
 	}
-	expected := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %d", testLogInfo.tableName, testLogInfo.npRef, testLogInfo.disposition,
-		testLogInfo.ofPriority, testLogInfo.srcIP, testLogInfo.srcPort, testLogInfo.destIP, testLogInfo.destPort,
+	expected := fmt.Sprintf("%s %s %s %s %s %s %s %s %s %s %d", testLogInfo.tableName, testLogInfo.npRef, testLogInfo.ruleName,
+		testLogInfo.disposition, testLogInfo.ofPriority, testLogInfo.srcIP, testLogInfo.srcPort, testLogInfo.destIP, testLogInfo.destPort,
 		testLogInfo.protocolStr, testLogInfo.pktLength)
 	return testLogInfo, expected
 }
@@ -220,3 +230,85 @@ func TestDropPacketMultiDedupLog(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, 1, c2)
 }
+
+func TestGetNetworkPolicyInfo(t *testing.T) {
+	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),
+		})
+	c := &Controller{ofClient: &openflowtest.MockClient{}}
+	ob := new(logInfo)
+	regID := openflow.APDispositionField.GetRegID()
+	dispositionMatch := openflow15.MatchField{
+		Class:   openflow15.OXM_CLASS_PACKET_REGS,
+		Field:   uint8(regID / 2),
+		HasMask: false,
+		Value:   &openflow15.ByteArrayField{Data: []byte{1, 1, 1, 1}},
+	}
+	matchers := []openflow15.MatchField{dispositionMatch}
+	pktIn := &ofctrl.PacketIn{TableId: 17, Match: openflow15.Match{Fields: matchers}}
+
+	err := getNetworkPolicyInfo(pktIn, c, ob)
+	assert.Equal(t, string(v1beta2.K8sNetworkPolicy), ob.npRef)
+	assert.Equal(t, "<nil>", ob.ofPriority)
+	assert.Equal(t, "<nil>", ob.ruleName)
+	require.NoError(t, err)
+}
+
+func TestGetPacketInfo(t *testing.T) {
+	tests := []struct {
+		name   string
+		packet *binding.Packet
+		ob     *logInfo
+		wantOb *logInfo
+	}{
+		{
+			name: "TCP packet",
+			packet: &binding.Packet{
+				SourceIP:        net.IPv4zero,
+				DestinationIP:   net.IPv4(1, 1, 1, 1),
+				IPLength:        60,
+				IPProto:         ip.TCPProtocol,
+				SourcePort:      35402,
+				DestinationPort: 80,
+			},
+			ob: new(logInfo),
+			wantOb: &logInfo{
+				srcIP:       "0.0.0.0",
+				srcPort:     "35402",
+				destIP:      "1.1.1.1",
+				destPort:    "80",
+				protocolStr: "TCP",
+				pktLength:   60,
+			},
+		},
+		{
+			name: "ICMP packet",
+			packet: &binding.Packet{
+				SourceIP:      net.IPv4zero,
+				DestinationIP: net.IPv4(1, 1, 1, 1),
+				IPLength:      60,
+				IPProto:       ip.ICMPProtocol,
+			},
+			ob: new(logInfo),
+			wantOb: &logInfo{
+				srcIP:       "0.0.0.0",
+				srcPort:     "<nil>",
+				destIP:      "1.1.1.1",
+				destPort:    "<nil>",
+				protocolStr: "ICMP",
+				pktLength:   60,
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		getPacketInfo(tc.packet, tc.ob)
+		assert.Equal(t, tc.wantOb, tc.ob)
+	}
+}
diff --git a/pkg/agent/controller/trafficcontrol/controller_test.go b/pkg/agent/controller/trafficcontrol/controller_test.go
index 59fc7da73..498851ed4 100644
--- a/pkg/agent/controller/trafficcontrol/controller_test.go
+++ b/pkg/agent/controller/trafficcontrol/controller_test.go
@@ -279,7 +279,7 @@ func generateTrafficControlState(direction v1alpha2.Direction,
 }
 
 func waitEvents(t *testing.T, expectedEvents int, c *fakeController) {
-	require.NoError(t, wait.PollImmediate(10*time.Millisecond, time.Second, func() (done bool, err error) {
+	require.NoError(t, wait.PollImmediate(10*time.Millisecond, 5*time.Second, func() (done bool, err error) {
 		return c.queue.Len() == expectedEvents, nil
 	}))
 }
diff --git a/pkg/agent/flowexporter/exporter/exporter_test.go b/pkg/agent/flowexporter/exporter/exporter_test.go
index 72328ea5c..78e9b2040 100644
--- a/pkg/agent/flowexporter/exporter/exporter_test.go
+++ b/pkg/agent/flowexporter/exporter/exporter_test.go
@@ -26,7 +26,6 @@ import (
 	ipfixentities "github.com/vmware/go-ipfix/pkg/entities"
 	ipfixentitiestesting "github.com/vmware/go-ipfix/pkg/entities/testing"
 	"github.com/vmware/go-ipfix/pkg/exporter"
-	"github.com/vmware/go-ipfix/pkg/registry"
 	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
 	"k8s.io/component-base/metrics/legacyregistry"
 
@@ -45,7 +44,7 @@ const (
 )
 
 func init() {
-	registry.LoadRegistry()
+	ipfixregistry.LoadRegistry()
 }
 
 func TestFlowExporter_sendTemplateSet(t *testing.T) {
@@ -389,11 +388,11 @@ func getConnection(isIPv6 bool, isPresent bool, statusFlag uint32, protoID uint8
 		DestinationPodName:            "",
 		IngressNetworkPolicyName:      "",
 		IngressNetworkPolicyNamespace: "",
-		IngressNetworkPolicyType:      registry.PolicyTypeK8sNetworkPolicy,
+		IngressNetworkPolicyType:      ipfixregistry.PolicyTypeK8sNetworkPolicy,
 		IngressNetworkPolicyRuleName:  "",
 		EgressNetworkPolicyName:       "np",
 		EgressNetworkPolicyNamespace:  "np-ns",
-		EgressNetworkPolicyType:       registry.PolicyTypeK8sNetworkPolicy,
+		EgressNetworkPolicyType:       ipfixregistry.PolicyTypeK8sNetworkPolicy,
 		EgressNetworkPolicyRuleName:   "",
 		DestinationServicePortName:    "service",
 		TCPState:                      tcpState,
diff --git a/pkg/agent/openflow/multicast_test.go b/pkg/agent/openflow/multicast_test.go
new file mode 100644
index 000000000..2a6eec7f7
--- /dev/null
+++ b/pkg/agent/openflow/multicast_test.go
@@ -0,0 +1,185 @@
+// Copyright 2022 Antrea Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package openflow
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+
+	"antrea.io/antrea/pkg/agent/config"
+	"antrea.io/antrea/pkg/agent/openflow/cookie"
+	oftest "antrea.io/antrea/pkg/agent/openflow/testing"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+	mocks "antrea.io/antrea/pkg/ovs/openflow/testing"
+)
+
+var (
+	bakMulticastEgressRuleTable       binding.Table
+	bakMulticastEgressPodMetricTable  binding.Table
+	bakMulticastRoutingTable          binding.Table
+	bakMulticastOutputTable           binding.Table
+	bakMulticastIngressPodMetricTable binding.Table
+
+	mockMulticastEgressRuleTable       *mocks.MockTable
+	mockMulticastEgressPodMetricTable  *mocks.MockTable
+	mockMulticastRoutingTable          *mocks.MockTable
+	mockMulticastOutputTable           *mocks.MockTable
+	mockMulticastIngressPodMetricTable *mocks.MockTable
+
+	mockFeatureMulticast = featureMulticast{
+		enableAntreaPolicy: true,
+		cookieAllocator:    cookie.NewAllocator(0),
+		category:           cookie.Multicast,
+		cachedFlows:        newFlowCategoryCache(),
+		groupCache:         sync.Map{},
+	}
+)
+
+func newMockPktInFlowBuilder(ctrl *gomock.Controller) *mocks.MockFlowBuilder {
+	pktInFlowBuilder := mocks.NewMockFlowBuilder(ctrl)
+	pktInFlowBuilder.EXPECT().Cookie(gomock.Any()).Return(pktInFlowBuilder).AnyTimes()
+	pktInFlowBuilder.EXPECT().MatchProtocol(gomock.Any()).Return(pktInFlowBuilder).AnyTimes()
+	pktInFlowBuilder.EXPECT().MatchRegMark(gomock.Any()).Return(pktInFlowBuilder).AnyTimes()
+	action := mocks.NewMockAction(ctrl)
+	action.EXPECT().LoadRegMark(gomock.Any()).Return(pktInFlowBuilder).AnyTimes()
+	pktInFlowBuilder.EXPECT().Action().Return(action).Times(1)
+	action.EXPECT().SendToController(gomock.Any()).Return(pktInFlowBuilder).AnyTimes()
+	pktInFlowBuilder.EXPECT().Action().Return(action).Times(1)
+
+	pktInFlow := mocks.NewMockFlow(ctrl)
+	pktInFlowBuilder.EXPECT().Done().Return(pktInFlow).Times(1)
+	pktInFlow.EXPECT().MatchString().Return("").AnyTimes()
+
+	return pktInFlowBuilder
+}
+
+func newMockExternalMulticastReceiver(ctrl *gomock.Controller) *mocks.MockFlowBuilder {
+	externalMcastReceiverFlowBuilder := mocks.NewMockFlowBuilder(ctrl)
+
+	externalMcastReceiverFlowBuilder.EXPECT().Cookie(gomock.Any()).Return(externalMcastReceiverFlowBuilder).AnyTimes()
+	externalMcastReceiverFlowBuilder.EXPECT().MatchProtocol(gomock.Any()).Return(externalMcastReceiverFlowBuilder).AnyTimes()
+
+	action := mocks.NewMockAction(ctrl)
+	action.EXPECT().LoadRegMark(gomock.Any()).Return(externalMcastReceiverFlowBuilder).AnyTimes()
+	externalMcastReceiverFlowBuilder.EXPECT().Action().Return(action).Times(1)
+	action.EXPECT().LoadToRegField(gomock.Any(), gomock.Any()).Return(externalMcastReceiverFlowBuilder).AnyTimes()
+	externalMcastReceiverFlowBuilder.EXPECT().Action().Return(action).Times(1)
+	action.EXPECT().GotoStage(gomock.Any()).Return(externalMcastReceiverFlowBuilder).AnyTimes()
+	externalMcastReceiverFlowBuilder.EXPECT().Action().Return(action).Times(1)
+
+	externalMcastReceiverFlow := mocks.NewMockFlow(ctrl)
+	externalMcastReceiverFlowBuilder.EXPECT().Done().Return(externalMcastReceiverFlow).Times(1)
+	externalMcastReceiverFlow.EXPECT().MatchString().Return("").AnyTimes()
+
+	return externalMcastReceiverFlowBuilder
+}
+
+func newMockMulticastSkipIGMPMetricFlows(ctrl *gomock.Controller) *mocks.MockFlowBuilder {
+	mockFlowBuilder := mocks.NewMockFlowBuilder(ctrl)
+	mockFlowBuilder.EXPECT().Cookie(gomock.Any()).Return(mockFlowBuilder).AnyTimes()
+	mockFlowBuilder.EXPECT().MatchProtocol(gomock.Any()).Return(mockFlowBuilder).AnyTimes()
+	action := mocks.NewMockAction(ctrl)
+	action.EXPECT().NextTable().Return(mockFlowBuilder).AnyTimes()
+	mockFlowBuilder.EXPECT().Action().Return(action).Times(1)
+	mockFlow := mocks.NewMockFlow(ctrl)
+	mockFlowBuilder.EXPECT().Done().Return(mockFlow).Times(1)
+	mockFlow.EXPECT().MatchString().Return("").AnyTimes()
+	return mockFlowBuilder
+}
+
+func newMockIGMPFlowBuilder(ctrl *gomock.Controller) *mocks.MockFlowBuilder {
+	igmpFlowBuilder := mocks.NewMockFlowBuilder(ctrl)
+	igmpFlowBuilder.EXPECT().Cookie(gomock.Any()).Return(igmpFlowBuilder).Times(1)
+	igmpFlowBuilder.EXPECT().MatchProtocol(gomock.Any()).Return(igmpFlowBuilder).Times(1)
+	igmpFlowBuilder.EXPECT().MatchRegMark(gomock.Any()).Return(igmpFlowBuilder).Times(1)
+	action := mocks.NewMockAction(ctrl)
+	action.EXPECT().GotoStage(gomock.Any()).Return(igmpFlowBuilder).Times(1)
+	igmpFlowBuilder.EXPECT().Action().Return(action).Times(1)
+	igmpFlow := mocks.NewMockFlow(ctrl)
+	igmpFlowBuilder.EXPECT().Done().Return(igmpFlow).Times(1)
+	igmpFlow.EXPECT().MatchString().Return("").AnyTimes()
+	return igmpFlowBuilder
+}
+
+func initMockTables(ctrl *gomock.Controller) {
+	mockMulticastEgressRuleTable = mocks.NewMockTable(ctrl)
+	mockMulticastEgressPodMetricTable = mocks.NewMockTable(ctrl)
+	mockMulticastRoutingTable = mocks.NewMockTable(ctrl)
+	mockMulticastOutputTable = mocks.NewMockTable(ctrl)
+	mockMulticastIngressPodMetricTable = mocks.NewMockTable(ctrl)
+
+	bakMulticastEgressRuleTable = MulticastEgressRuleTable.ofTable
+	bakMulticastEgressPodMetricTable = MulticastEgressPodMetricTable.ofTable
+	bakMulticastRoutingTable = MulticastRoutingTable.ofTable
+	bakMulticastOutputTable = MulticastOutputTable.ofTable
+	bakMulticastIngressPodMetricTable = MulticastIngressPodMetricTable.ofTable
+
+	MulticastEgressRuleTable.ofTable = mockMulticastEgressRuleTable
+	MulticastEgressPodMetricTable.ofTable = mockMulticastEgressPodMetricTable
+	MulticastRoutingTable.ofTable = mockMulticastRoutingTable
+	MulticastOutputTable.ofTable = mockMulticastOutputTable
+	MulticastIngressPodMetricTable.ofTable = mockMulticastIngressPodMetricTable
+}
+
+func resetMulticastTables() {
+	MulticastEgressRuleTable.ofTable = bakMulticastEgressRuleTable
+	MulticastEgressPodMetricTable.ofTable = bakMulticastEgressPodMetricTable
+	MulticastRoutingTable.ofTable = bakMulticastRoutingTable
+	MulticastOutputTable.ofTable = bakMulticastOutputTable
+	MulticastIngressPodMetricTable.ofTable = bakMulticastIngressPodMetricTable
+}
+
+func newMcastClient(ctrl *gomock.Controller) *client {
+	bridge := mocks.NewMockBridge(ctrl)
+	bridge.EXPECT().AddFlowsInBundle(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	m := oftest.NewMockOFEntryOperations(ctrl)
+	m.EXPECT().AddAll(gomock.Any()).Return(nil).AnyTimes()
+	return &client{
+		bridge:             bridge,
+		ipProtocols:        []binding.Protocol{binding.ProtocolIP},
+		cookieAllocator:    cookie.NewAllocator(0),
+		ofEntryOperations:  m,
+		featureMulticast:   &mockFeatureMulticast,
+		enableMulticast:    true,
+		enableAntreaPolicy: true,
+	}
+}
+
+func TestInstallMulticastInitialFlows(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	c = newMcastClient(ctrl)
+	initMockTables(ctrl)
+	defer resetMulticastTables()
+
+	c.enableMulticast = true
+	c.enableAntreaPolicy = true
+	c.featureMulticast = &mockFeatureMulticast
+	mockMulticastRoutingTable.EXPECT().BuildFlow(gomock.Any()).Return(newMockPktInFlowBuilder(ctrl)).Times(1)
+	mockMulticastRoutingTable.EXPECT().BuildFlow(gomock.Any()).Return(newMockExternalMulticastReceiver(ctrl)).Times(1)
+	mockMulticastEgressPodMetricTable.EXPECT().BuildFlow(gomock.Any()).Return(newMockMulticastSkipIGMPMetricFlows(ctrl)).Times(1)
+	mockMulticastIngressPodMetricTable.EXPECT().BuildFlow(gomock.Any()).Return(newMockMulticastSkipIGMPMetricFlows(ctrl)).Times(1)
+	mockMulticastEgressRuleTable.EXPECT().BuildFlow(priorityTopAntreaPolicy).Return(newMockIGMPFlowBuilder(ctrl)).Times(1)
+	c.nodeConfig = &config.NodeConfig{PodIPv4CIDR: podIPv4CIDR, PodIPv6CIDR: nil}
+	c.networkConfig = &config.NetworkConfig{}
+	c.pipelines = pipelineMap
+
+	err := c.InstallMulticastInitialFlows(uint8(PacketInReasonMC))
+	require.NoErrorf(t, err, "Failed to install multicast initial flows")
+}
diff --git a/pkg/agent/openflow/network_policy_test.go b/pkg/agent/openflow/network_policy_test.go
index f8a5b3d93..c22269587 100644
--- a/pkg/agent/openflow/network_policy_test.go
+++ b/pkg/agent/openflow/network_policy_test.go
@@ -1263,3 +1263,71 @@ func setMockOFTables(ctrl *gomock.Controller, tableMap map[*Table]**mocks.MockTa
 		*mockTable = t // Update the value with generated mock table.
 	}
 }
+
+func TestClient_GetPolicyInfoFromConjunction(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	preparePipelines()
+	c = prepareClient(ctrl, false)
+
+	ruleID1 := uint32(101)
+	ruleID2 := uint32(102)
+	npRef := &v1beta2.NetworkPolicyReference{
+		Type:      v1beta2.K8sNetworkPolicy,
+		Namespace: "ns1",
+		Name:      "np1",
+		UID:       "id1",
+	}
+	conj1 := &policyRuleConjunction{
+		id:       ruleID1,
+		npRef:    npRef,
+		ruleName: fmt.Sprint(ruleID1),
+	}
+	conj2 := &policyRuleConjunction{
+		id:          ruleID2,
+		actionFlows: []binding.Flow{EgressRuleTable.ofTable.BuildFlow(priority100).Done()},
+		npRef:       npRef,
+		ruleName:    fmt.Sprint(ruleID2),
+	}
+	c.featureNetworkPolicy.policyCache.Add(conj1)
+	c.featureNetworkPolicy.policyCache.Add(conj2)
+
+	tests := []struct {
+		name         string
+		ruleID       uint32
+		wantNpRef    string
+		wantPriority string
+		wantRuleName string
+	}{
+		{
+			name:         "conjunction not found",
+			ruleID:       uint32(100),
+			wantNpRef:    "",
+			wantPriority: "",
+			wantRuleName: "",
+		},
+		{
+			name:         "conjunction empty priorities",
+			ruleID:       ruleID1,
+			wantNpRef:    "",
+			wantPriority: "",
+			wantRuleName: "",
+		},
+		{
+			name:         "conjunction no error",
+			ruleID:       ruleID2,
+			wantNpRef:    "K8sNetworkPolicy:ns1/np1",
+			wantPriority: "100",
+			wantRuleName: fmt.Sprint(ruleID2),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			gotNpRef, gotPriority, gotRuleName := c.GetPolicyInfoFromConjunction(tc.ruleID)
+			assert.Equal(t, tc.wantNpRef, gotNpRef)
+			assert.Equal(t, tc.wantPriority, gotPriority)
+			assert.Equal(t, tc.wantRuleName, gotRuleName)
+		})
+	}
+}
diff --git a/pkg/agent/openflow/pipeline_test.go b/pkg/agent/openflow/pipeline_test.go
index 2285ee27b..12bbb5a40 100644
--- a/pkg/agent/openflow/pipeline_test.go
+++ b/pkg/agent/openflow/pipeline_test.go
@@ -262,11 +262,13 @@ func TestBuildPipeline(t *testing.T) {
 				pipelineMulticast: {
 					MulticastEgressRuleTable,
 					MulticastEgressMetricTable,
+					MulticastEgressPodMetricTable,
 
 					MulticastRoutingTable,
 
 					MulticastIngressRuleTable,
 					MulticastIngressMetricTable,
+					MulticastIngressPodMetricTable,
 
 					MulticastOutputTable,
 				},
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index be27e495a..393728772 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -197,12 +197,13 @@ func (mr *MockClientMockRecorder) GetPodFlowKeys(arg0 interface{}) *gomock.Call
 }
 
 // GetPolicyInfoFromConjunction mocks base method
-func (m *MockClient) GetPolicyInfoFromConjunction(arg0 uint32) (string, string) {
+func (m *MockClient) GetPolicyInfoFromConjunction(arg0 uint32) (string, string, string) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetPolicyInfoFromConjunction", arg0)
 	ret0, _ := ret[0].(string)
 	ret1, _ := ret[1].(string)
-	return ret0, ret1
+	ret2, _ := ret[2].(string)
+	return ret0, ret1, ret2
 }
 
 // GetPolicyInfoFromConjunction indicates an expected call of GetPolicyInfoFromConjunction
diff --git a/pkg/controller/egress/controller_test.go b/pkg/controller/egress/controller_test.go
index d5880ced7..0ec499208 100644
--- a/pkg/controller/egress/controller_test.go
+++ b/pkg/controller/egress/controller_test.go
@@ -36,7 +36,6 @@ import (
 
 	"antrea.io/antrea/pkg/apis/controlplane"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
-	corev1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	"antrea.io/antrea/pkg/client/clientset/versioned"
 	fakeversioned "antrea.io/antrea/pkg/client/clientset/versioned/fake"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions"
@@ -69,7 +68,7 @@ func newEgress(name, egressIP, externalIPPool string, podSelector, namespaceSele
 	egress := &v1alpha2.Egress{
 		ObjectMeta: metav1.ObjectMeta{Name: name},
 		Spec: v1alpha2.EgressSpec{
-			AppliedTo: corev1a2.AppliedTo{
+			AppliedTo: v1alpha2.AppliedTo{
 				PodSelector:       podSelector,
 				NamespaceSelector: namespaceSelector,
 			},
@@ -87,10 +86,10 @@ func newExternalIPPool(name, cidr, start, end string) *v1alpha2.ExternalIPPool {
 		},
 	}
 	if len(cidr) > 0 {
-		pool.Spec.IPRanges = append(pool.Spec.IPRanges, corev1a2.IPRange{CIDR: cidr})
+		pool.Spec.IPRanges = append(pool.Spec.IPRanges, v1alpha2.IPRange{CIDR: cidr})
 	}
 	if len(start) > 0 && len(end) > 0 {
-		pool.Spec.IPRanges = append(pool.Spec.IPRanges, corev1a2.IPRange{Start: start, End: end})
+		pool.Spec.IPRanges = append(pool.Spec.IPRanges, v1alpha2.IPRange{Start: start, End: end})
 	}
 	return pool
 }
@@ -171,7 +170,7 @@ func TestAddEgress(t *testing.T) {
 			inputEgress: &v1alpha2.Egress{
 				ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 				Spec: v1alpha2.EgressSpec{
-					AppliedTo: corev1a2.AppliedTo{
+					AppliedTo: v1alpha2.AppliedTo{
 						PodSelector: &metav1.LabelSelector{
 							MatchLabels: map[string]string{"app": "foo"},
 						},
@@ -205,7 +204,7 @@ func TestAddEgress(t *testing.T) {
 			inputEgress: &v1alpha2.Egress{
 				ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 				Spec: v1alpha2.EgressSpec{
-					AppliedTo: corev1a2.AppliedTo{
+					AppliedTo: v1alpha2.AppliedTo{
 						NamespaceSelector: &metav1.LabelSelector{
 							MatchLabels: nsDefault.Labels,
 						},
@@ -237,7 +236,7 @@ func TestAddEgress(t *testing.T) {
 			inputEgress: &v1alpha2.Egress{
 				ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 				Spec: v1alpha2.EgressSpec{
-					AppliedTo: corev1a2.AppliedTo{
+					AppliedTo: v1alpha2.AppliedTo{
 						PodSelector: &metav1.LabelSelector{
 							MatchLabels: map[string]string{"app": "foo"},
 						},
@@ -269,7 +268,7 @@ func TestAddEgress(t *testing.T) {
 			inputEgress: &v1alpha2.Egress{
 				ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 				Spec: v1alpha2.EgressSpec{
-					AppliedTo: corev1a2.AppliedTo{
+					AppliedTo: v1alpha2.AppliedTo{
 						PodSelector: &metav1.LabelSelector{
 							MatchLabels: map[string]string{"app": "foo"},
 						},
@@ -374,7 +373,7 @@ func TestUpdateEgress(t *testing.T) {
 	egress := &v1alpha2.Egress{
 		ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 		Spec: v1alpha2.EgressSpec{
-			AppliedTo: corev1a2.AppliedTo{
+			AppliedTo: v1alpha2.AppliedTo{
 				PodSelector: &metav1.LabelSelector{
 					MatchLabels: map[string]string{"app": "foo"},
 				},
@@ -449,7 +448,7 @@ func TestUpdateEgress(t *testing.T) {
 	}, getEvent())
 
 	// Updating the Egress's spec to make it match no Pods on this Node and use a new ExternalIPPool.
-	egress.Spec.AppliedTo = corev1a2.AppliedTo{
+	egress.Spec.AppliedTo = v1alpha2.AppliedTo{
 		PodSelector: &metav1.LabelSelector{
 			MatchLabels: map[string]string{"app": "non-existing-app"},
 		},
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index 60979921d..7c9ea1aaf 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -2557,12 +2557,14 @@ func testANPMultipleAppliedTo(t *testing.T, data *TestData, singleRule bool) {
 
 // testAuditLoggingBasic tests that audit logs are generated when egress drop applied
 func testAuditLoggingBasic(t *testing.T, data *TestData) {
+	npRef := "test-log-acnp-deny"
+	ruleName := "DropToZ"
 	builder := &ClusterNetworkPolicySpecBuilder{}
-	builder = builder.SetName("test-log-acnp-deny").
+	builder = builder.SetName(npRef).
 		SetPriority(1.0).
 		SetAppliedToGroup([]ACNPAppliedToSpec{{PodSelector: map[string]string{"pod": "a"}, NSSelector: map[string]string{"ns": namespaces["x"]}}})
 	builder.AddEgress(ProtocolTCP, &p80, nil, nil, nil, nil, nil, nil, nil, nil, map[string]string{"ns": namespaces["z"]},
-		nil, nil, false, nil, crdv1alpha1.RuleActionDrop, "", "", nil)
+		nil, nil, false, nil, crdv1alpha1.RuleActionDrop, "", ruleName, nil)
 	builder.AddEgressLogging()
 
 	acnp, err := k8sUtils.CreateOrUpdateACNP(builder.Get())
@@ -2621,7 +2623,7 @@ func testAuditLoggingBasic(t *testing.T, data *TestData) {
 					}
 					expectedNumEntries += 1
 					// The audit log should contain log entry `... Drop <ofPriority> <x/a IP> <z/* IP> ...`
-					re := regexp.MustCompile(`Drop [0-9]+ ` + srcIPs[i] + ` [0-9]+ ` + dstIPs[j] + ` ` + strconv.Itoa(int(p80)))
+					re := regexp.MustCompile(npRef + ` ` + ruleName + ` Drop [0-9]+ ` + srcIPs[i] + ` [0-9]+ ` + dstIPs[j] + ` ` + strconv.Itoa(int(p80)))
 					if re.MatchString(stdout) {
 						actualNumEntries += 1
 					} else {
@@ -2649,8 +2651,9 @@ func testAuditLoggingEnableNP(t *testing.T, data *TestData) {
 	failOnError(data.updateNamespaceWithAnnotations(namespaces["x"], map[string]string{networkpolicy.EnableNPLoggingAnnotationKey: "true"}), t)
 	// Add a K8s namespaced NetworkPolicy in ns x that allow ingress traffic from
 	// Pod x/b to x/a which default denies other ingress including from Pod x/c to x/a
+	npRef := "allow-x-b-to-x-a"
 	k8sNPBuilder := &NetworkPolicySpecBuilder{}
-	k8sNPBuilder = k8sNPBuilder.SetName(namespaces["x"], "allow-x-b-to-x-a").
+	k8sNPBuilder = k8sNPBuilder.SetName(namespaces["x"], npRef).
 		SetPodSelector(map[string]string{"pod": "a"}).
 		SetTypeIngress().
 		AddIngress(v1.ProtocolTCP, &p80, nil, nil, nil,
@@ -2699,7 +2702,7 @@ func testAuditLoggingEnableNP(t *testing.T, data *TestData) {
 
 		var expectedNumEntries, actualNumEntries int
 		srcPods := []string{namespaces["x"] + "/b", namespaces["x"] + "/c"}
-		expectedLogPrefix := []string{"allow-x-b-to-x-a Allow [0-9]+ ", "K8sNetworkPolicy Drop -1 "}
+		expectedLogPrefix := []string{npRef + " <nil> Allow [0-9]+ ", "K8sNetworkPolicy <nil> Drop <nil> "}
 		destIPs, _ := podIPs[namespaces["x"]+"/a"]
 		for i := 0; i < len(srcPods); i++ {
 			srcIPs, _ := podIPs[srcPods[i]]
diff --git a/test/e2e/providers/vagrant.go b/test/e2e/providers/vagrant.go
index e1d546a5f..ea90ac36f 100644
--- a/test/e2e/providers/vagrant.go
+++ b/test/e2e/providers/vagrant.go
@@ -139,7 +139,7 @@ func (provider *VagrantProvider) GetKubeconfigPath() (string, error) {
 	}
 	kubeconfigPath := path.Join(vagrantPath, "playbook", "kube", "config")
 	if _, err := os.Stat(kubeconfigPath); os.IsNotExist(err) {
-		return "", fmt.Errorf("Kubeconfig file not found at expected location '%s'", kubeconfigPath)
+		return "", fmt.Errorf("kubeconfig file not found at expected location '%s'", kubeconfigPath)
 	}
 	return kubeconfigPath, nil
 }
diff --git a/test/e2e/service_externalip_test.go b/test/e2e/service_externalip_test.go
index a639bb808..40635e36f 100644
--- a/test/e2e/service_externalip_test.go
+++ b/test/e2e/service_externalip_test.go
@@ -26,7 +26,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -215,7 +214,7 @@ func testServiceExternalTrafficPolicyLocal(t *testing.T, data *TestData) {
 				antreaagenttypes.ServiceExternalIPPoolAnnotationKey: ipPool.Name,
 			}
 			service, err = data.CreateServiceWithAnnotations(fmt.Sprintf("test-svc-local-%d", idx),
-				data.testNamespace, 80, 80, corev1.ProtocolTCP, nil, false, true, v1.ServiceTypeLoadBalancer, nil, annotation)
+				data.testNamespace, 80, 80, v1.ProtocolTCP, nil, false, true, v1.ServiceTypeLoadBalancer, nil, annotation)
 			require.NoError(t, err)
 			defer data.clientset.CoreV1().Services(service.Namespace).Delete(context.TODO(), service.Name, metav1.DeleteOptions{})
 
@@ -332,7 +331,7 @@ func testServiceWithExternalIPCRUD(t *testing.T, data *TestData) {
 				antreaagenttypes.ServiceExternalIPPoolAnnotationKey: ipPool.Name,
 			}
 			service, err = data.CreateServiceWithAnnotations(fmt.Sprintf("test-svc-eip-%d", idx),
-				data.testNamespace, 80, 80, corev1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
+				data.testNamespace, 80, 80, v1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
 			require.NoError(t, err)
 
 			defer data.clientset.CoreV1().Services(service.Namespace).Delete(context.TODO(), service.Name, metav1.DeleteOptions{})
@@ -424,7 +423,7 @@ func testServiceUpdateExternalIP(t *testing.T, data *TestData) {
 				antreaagenttypes.ServiceExternalIPPoolAnnotationKey: originalPool.Name,
 			}
 			service, err := data.CreateServiceWithAnnotations(fmt.Sprintf("test-update-eip-%d", idx),
-				data.testNamespace, 80, 80, corev1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
+				data.testNamespace, 80, 80, v1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
 			require.NoError(t, err)
 			defer data.clientset.CoreV1().Services(service.Namespace).Delete(context.TODO(), service.Name, metav1.DeleteOptions{})
 
@@ -506,7 +505,7 @@ func testServiceNodeFailure(t *testing.T, data *TestData) {
 				antreaagenttypes.ServiceExternalIPPoolAnnotationKey: externalIPPoolTwoNodes.Name,
 			}
 			service, err := data.CreateServiceWithAnnotations("test-service-node-failure", data.testNamespace, 80, 80,
-				corev1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
+				v1.ProtocolTCP, nil, false, false, v1.ServiceTypeLoadBalancer, nil, annotation)
 			require.NoError(t, err)
 			defer data.clientset.CoreV1().Services(service.Namespace).Delete(context.TODO(), service.Name, metav1.DeleteOptions{})
 
@@ -565,10 +564,10 @@ func testExternalIPAccess(t *testing.T, data *TestData) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ipFamily := corev1.IPv4Protocol
+			ipFamily := v1.IPv4Protocol
 			if utilnet.IsIPv6CIDRString(tt.externalIPCIDR) {
 				skipIfNotIPv6Cluster(t)
-				ipFamily = corev1.IPv6Protocol
+				ipFamily = v1.IPv6Protocol
 			} else {
 				skipIfNotIPv4Cluster(t)
 			}
@@ -624,7 +623,7 @@ func testExternalIPAccess(t *testing.T, data *TestData) {
 					annotations := map[string]string{
 						antreaagenttypes.ServiceExternalIPPoolAnnotationKey: ipPool.Name,
 					}
-					service, err := data.CreateServiceWithAnnotations(et.serviceName, data.testNamespace, port, port, corev1.ProtocolTCP, map[string]string{"app": "agnhost"}, false, et.externalTrafficPolicyLocal, corev1.ServiceTypeLoadBalancer, &ipFamily, annotations)
+					service, err := data.CreateServiceWithAnnotations(et.serviceName, data.testNamespace, port, port, v1.ProtocolTCP, map[string]string{"app": "agnhost"}, false, et.externalTrafficPolicyLocal, v1.ServiceTypeLoadBalancer, &ipFamily, annotations)
 					require.NoError(t, err)
 					defer data.deleteService(service.Namespace, service.Name)
 
@@ -664,8 +663,8 @@ sleep 3600`, tt.clientName, tt.clientIP, tt.localIP, tt.clientIPMaskLen)
 
 					_, err = data.PodWaitFor(defaultTimeout, tt.clientName, data.testNamespace, func(p *v1.Pod) (bool, error) {
 						for _, condition := range p.Status.Conditions {
-							if condition.Type == corev1.PodReady {
-								return condition.Status == corev1.ConditionTrue, nil
+							if condition.Type == v1.PodReady {
+								return condition.Status == v1.ConditionTrue, nil
 							}
 						}
 						return false, nil
@@ -722,7 +721,7 @@ func (data *TestData) getServiceAssignedNode(node string, service *v1.Service) (
 	return serviceExternalIPInfo[0].AssignedNode, nil
 }
 
-func (data *TestData) waitForServiceConfigured(service *v1.Service, expectedExternalIP string, expectedNodeName string) (*corev1.Service, string, error) {
+func (data *TestData) waitForServiceConfigured(service *v1.Service, expectedExternalIP string, expectedNodeName string) (*v1.Service, string, error) {
 	var assignedNode string
 	err := wait.PollImmediate(200*time.Millisecond, 15*time.Second, func() (done bool, err error) {
 		service, err = data.clientset.CoreV1().Services(service.Namespace).Get(context.TODO(), service.Name, metav1.GetOptions{})