Repository: antrea-io/antrea
Pull request #2569 (2021-08-10 08:40:13)

[CI][Windows] Refactor log collection
* Collect kubelet and docker logs on Windows VMs
* Refactor code

Signed-off-by: Zhecheng Li <lzhecheng@vmware.com>

lzhecheng: /test-windows-networkpolicy
lzhecheng: /skip-all
lzhecheng: @antoninbas please help merge the PR.

Patch:
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index ff921c05a..f773f0aa3 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -38,7 +38,36 @@ spec:
     singular: antreaagentinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of this Agent
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of this Agent
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Node on which this Agent is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of local Pods managed by this Agent
+      jsonPath: .localPodNum
+      name: Num Pods
+      priority: 2
+      type: integer
+    - description: Subnets used by this Agent for Pod IPAM
+      jsonPath: .nodeSubnets
+      name: Subnets
+      priority: 2
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
@@ -86,7 +115,36 @@ spec:
     singular: antreacontrollerinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of the Controller
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of the Controller
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Number of Agents connected to the Controller
+      jsonPath: .connectedAgentNum
+      name: Connected Agents
+      priority: 1
+      type: integer
+    - description: Node on which the Controller is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of Network Policies computed by Controller
+      jsonPath: .networkPolicyControllerInfo.networkPolicyNum
+      name: Num Network Policies
+      priority: 2
+      type: integer
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index f601939e6..0eda363b5 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -38,7 +38,36 @@ spec:
     singular: antreaagentinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of this Agent
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of this Agent
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Node on which this Agent is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of local Pods managed by this Agent
+      jsonPath: .localPodNum
+      name: Num Pods
+      priority: 2
+      type: integer
+    - description: Subnets used by this Agent for Pod IPAM
+      jsonPath: .nodeSubnets
+      name: Subnets
+      priority: 2
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
@@ -86,7 +115,36 @@ spec:
     singular: antreacontrollerinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of the Controller
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of the Controller
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Number of Agents connected to the Controller
+      jsonPath: .connectedAgentNum
+      name: Connected Agents
+      priority: 1
+      type: integer
+    - description: Node on which the Controller is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of Network Policies computed by Controller
+      jsonPath: .networkPolicyControllerInfo.networkPolicyNum
+      name: Num Network Policies
+      priority: 2
+      type: integer
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index b7454d715..811d38195 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -38,7 +38,36 @@ spec:
     singular: antreaagentinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of this Agent
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of this Agent
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Node on which this Agent is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of local Pods managed by this Agent
+      jsonPath: .localPodNum
+      name: Num Pods
+      priority: 2
+      type: integer
+    - description: Subnets used by this Agent for Pod IPAM
+      jsonPath: .nodeSubnets
+      name: Subnets
+      priority: 2
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
@@ -86,7 +115,36 @@ spec:
     singular: antreacontrollerinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of the Controller
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of the Controller
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Number of Agents connected to the Controller
+      jsonPath: .connectedAgentNum
+      name: Connected Agents
+      priority: 1
+      type: integer
+    - description: Node on which the Controller is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of Network Policies computed by Controller
+      jsonPath: .networkPolicyControllerInfo.networkPolicyNum
+      name: Num Network Policies
+      priority: 2
+      type: integer
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index 7f2ed3ffd..0efca5c4b 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -38,7 +38,36 @@ spec:
     singular: antreaagentinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of this Agent
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of this Agent
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Node on which this Agent is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of local Pods managed by this Agent
+      jsonPath: .localPodNum
+      name: Num Pods
+      priority: 2
+      type: integer
+    - description: Subnets used by this Agent for Pod IPAM
+      jsonPath: .nodeSubnets
+      name: Subnets
+      priority: 2
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
@@ -86,7 +115,36 @@ spec:
     singular: antreacontrollerinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of the Controller
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of the Controller
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Number of Agents connected to the Controller
+      jsonPath: .connectedAgentNum
+      name: Connected Agents
+      priority: 1
+      type: integer
+    - description: Node on which the Controller is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of Network Policies computed by Controller
+      jsonPath: .networkPolicyControllerInfo.networkPolicyNum
+      name: Num Network Policies
+      priority: 2
+      type: integer
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 86559f0a9..a39fb7381 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -38,7 +38,36 @@ spec:
     singular: antreaagentinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of this Agent
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of this Agent
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Node on which this Agent is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of local Pods managed by this Agent
+      jsonPath: .localPodNum
+      name: Num Pods
+      priority: 2
+      type: integer
+    - description: Subnets used by this Agent for Pod IPAM
+      jsonPath: .nodeSubnets
+      name: Subnets
+      priority: 2
+      type: string
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
@@ -86,7 +115,36 @@ spec:
     singular: antreacontrollerinfo
   scope: Cluster
   versions:
-  - name: v1beta1
+  - additionalPrinterColumns:
+    - description: Health status of the Controller
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].status
+      name: Healthy
+      type: string
+    - description: Last time the Healthy Condition was updated
+      jsonPath: .controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime
+      name: Last Heartbeat
+      type: date
+    - description: Version of the Controller
+      jsonPath: .version
+      name: Version
+      priority: 1
+      type: string
+    - description: Number of Agents connected to the Controller
+      jsonPath: .connectedAgentNum
+      name: Connected Agents
+      priority: 1
+      type: integer
+    - description: Node on which the Controller is running
+      jsonPath: .nodeRef.name
+      name: Node
+      priority: 1
+      type: string
+    - description: Number of Network Policies computed by Controller
+      jsonPath: .networkPolicyControllerInfo.networkPolicyNum
+      name: Num Network Policies
+      priority: 2
+      type: integer
+    name: v1beta1
     schema:
       openAPIV3Schema:
         type: object
diff --git a/build/yamls/base/crds.yml b/build/yamls/base/crds.yml
index 86cb22019..8bb85fb8b 100644
--- a/build/yamls/base/crds.yml
+++ b/build/yamls/base/crds.yml
@@ -199,6 +199,35 @@ spec:
         openAPIV3Schema:
           type: object
           x-kubernetes-preserve-unknown-fields: true
+      additionalPrinterColumns:
+        - description: Health status of the Controller
+          jsonPath: ".controllerConditions[?(@.type=='ControllerHealthy')].status"
+          name: Healthy
+          type: string
+        - description: Last time the Healthy Condition was updated
+          jsonPath: ".controllerConditions[?(@.type=='ControllerHealthy')].lastHeartbeatTime"
+          name: Last Heartbeat
+          type: date
+        - description: Version of the Controller
+          jsonPath: ".version"
+          name: Version
+          type: string
+          priority: 1
+        - description: Number of Agents connected to the Controller
+          jsonPath: ".connectedAgentNum"
+          name: Connected Agents
+          type: integer
+          priority: 1
+        - description: Node on which the Controller is running
+          jsonPath: ".nodeRef.name"
+          name: Node
+          type: string
+          priority: 1
+        - description: Number of Network Policies computed by Controller
+          jsonPath: ".networkPolicyControllerInfo.networkPolicyNum"
+          name: Num Network Policies
+          type: integer
+          priority: 2
   scope: Cluster
   names:
     plural: antreacontrollerinfos
@@ -221,6 +250,35 @@ spec:
         openAPIV3Schema:
           type: object
           x-kubernetes-preserve-unknown-fields: true
+      additionalPrinterColumns:
+        - description: Health status of this Agent
+          jsonPath: ".agentConditions[?(@.type=='AgentHealthy')].status"
+          name: Healthy
+          type: string
+        - description: Last time the Healthy Condition was updated
+          jsonPath: ".agentConditions[?(@.type=='AgentHealthy')].lastHeartbeatTime"
+          name: Last Heartbeat
+          type: date
+        - description: Version of this Agent
+          jsonPath: ".version"
+          name: Version
+          type: string
+          priority: 1
+        - description: Node on which this Agent is running
+          jsonPath: ".nodeRef.name"
+          name: Node
+          type: string
+          priority: 1
+        - description: Number of local Pods managed by this Agent
+          jsonPath: ".localPodNum"
+          name: Num Pods
+          type: integer
+          priority: 2
+        - description: Subnets used by this Agent for Pod IPAM
+          jsonPath: ".nodeSubnets"
+          name: Subnets
+          type: string
+          priority: 2
   scope: Cluster
   names:
     plural: antreaagentinfos
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 97499e41b..11a7fc2cb 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -197,7 +197,6 @@ func run(o *Options) error {
 	// In Antrea agent, status manager and audit logging will automatically be enabled
 	// if AntreaPolicy feature is enabled.
 	statusManagerEnabled := antreaPolicyEnabled
-	loggingEnabled := antreaPolicyEnabled
 
 	var denyConnStore *connections.DenyConnectionStore
 	if features.DefaultFeatureGate.Enabled(features.FlowExporter) {
@@ -212,7 +211,6 @@ func run(o *Options) error {
 		entityUpdates,
 		antreaPolicyEnabled,
 		statusManagerEnabled,
-		loggingEnabled,
 		denyConnStore,
 		asyncRuleDeleteInterval)
 	if err != nil {
diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go
index e6925ee46..de1893c05 100644
--- a/pkg/agent/agent.go
+++ b/pkg/agent/agent.go
@@ -41,7 +41,9 @@ import (
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
+	"antrea.io/antrea/pkg/features"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/ovs/ovsctl"
 	"antrea.io/antrea/pkg/util/env"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -129,6 +131,10 @@ func (i *Initializer) setupOVSBridge() error {
 		return err
 	}
 
+	if err := i.validateSupportedDPFeatures(); err != nil {
+		return err
+	}
+
 	if err := i.prepareOVSBridge(); err != nil {
 		return err
 	}
@@ -150,6 +156,35 @@ func (i *Initializer) setupOVSBridge() error {
 	return nil
 }
 
+func (i *Initializer) validateSupportedDPFeatures() error {
+	gotFeatures, err := ovsctl.NewClient(i.ovsBridge).GetDPFeatures()
+	if err != nil {
+		return err
+	}
+	// Basic requirements.
+	requiredFeatures := []ovsctl.DPFeature{
+		ovsctl.CTStateFeature,
+		ovsctl.CTZoneFeature,
+		ovsctl.CTMarkFeature,
+		ovsctl.CTLabelFeature,
+	}
+	// AntreaProxy requires CTStateNAT feature.
+	if features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
+		requiredFeatures = append(requiredFeatures, ovsctl.CTStateNATFeature)
+	}
+
+	for _, feature := range requiredFeatures {
+		supported, found := gotFeatures[feature]
+		if !found {
+			return fmt.Errorf("the required OVS DP feature '%s' support is unknown", feature)
+		}
+		if !supported {
+			return fmt.Errorf("the required OVS DP feature '%s' is not supported", feature)
+		}
+	}
+	return nil
+}
+
 // initInterfaceStore initializes InterfaceStore with all OVS ports retrieved
 // from the OVS bridge.
 func (i *Initializer) initInterfaceStore() error {
diff --git a/pkg/agent/controller/networkpolicy/audit_logging.go b/pkg/agent/controller/networkpolicy/audit_logging.go
new file mode 100644
index 000000000..e28c3f60a
--- /dev/null
+++ b/pkg/agent/controller/networkpolicy/audit_logging.go
@@ -0,0 +1,152 @@
+// Copyright 2021 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 networkpolicy
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"gopkg.in/natefinch/lumberjack.v2"
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/pkg/agent/openflow"
+	"antrea.io/antrea/pkg/util/logdir"
+)
+
+const (
+	logfileSubdir string = "networkpolicy"
+	logfileName   string = "np.log"
+)
+
+// AntreaPolicyLogger is used for Antrea policy audit logging.
+// Includes a lumberjack logger and a map used for log deduplication.
+type AntreaPolicyLogger struct {
+	anpLogger        *log.Logger
+	logDeduplication logRecordDedupMap
+}
+
+// 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
+	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
+	destIP      string // destination IP of the traffic logged
+	pktLength   uint16 // packet length of packetin
+	protocolStr string // protocol of the traffic logged
+}
+
+// logDedupRecord will be used as 1 sec buffer for log deduplication.
+type logDedupRecord struct {
+	count       int64       // record count of duplicate log
+	initTime    time.Time   // initial time upon receiving packet log
+	bufferTimer *time.Timer // 1 sec buffer for each log
+}
+
+// logRecordDedupMap includes a map of log buffers and a r/w mutex for accessing the map.
+type logRecordDedupMap struct {
+	logMutex sync.Mutex
+	logMap   map[string]*logDedupRecord
+}
+
+// getLogKey returns the log record in logDeduplication map by logMsg.
+func (a *AntreaPolicyLogger) getLogKey(logMsg string) *logDedupRecord {
+	a.logDeduplication.logMutex.Lock()
+	defer a.logDeduplication.logMutex.Unlock()
+	return a.logDeduplication.logMap[logMsg]
+}
+
+// logAfterTimer runs concurrently until buffer timer stops, then call terminateLogKey.
+func (a *AntreaPolicyLogger) logAfterTimer(logMsg string) {
+	logRecordTimer := a.getLogKey(logMsg).bufferTimer
+	<-logRecordTimer.C
+	a.terminateLogKey(logMsg)
+}
+
+// terminateLogKey logs and deletes the log record in logDeduplication map by logMsg.
+func (a *AntreaPolicyLogger) terminateLogKey(logMsg string) {
+	a.logDeduplication.logMutex.Lock()
+	defer a.logDeduplication.logMutex.Unlock()
+	if a.logDeduplication.logMap[logMsg].count == 1 {
+		a.anpLogger.Printf(logMsg)
+	} else {
+		a.anpLogger.Printf("%s [%d packets in %s]", logMsg, a.logDeduplication.logMap[logMsg].count, time.Since(a.logDeduplication.logMap[logMsg].initTime))
+	}
+	delete(a.logDeduplication.logMap, logMsg)
+}
+
+// updateLogKey initiates record or increases the count in logDeduplication corresponding to given logMsg.
+func (a *AntreaPolicyLogger) updateLogKey(logMsg string, bufferLength time.Duration) bool {
+	a.logDeduplication.logMutex.Lock()
+	defer a.logDeduplication.logMutex.Unlock()
+	_, exists := a.logDeduplication.logMap[logMsg]
+	if exists {
+		a.logDeduplication.logMap[logMsg].count++
+	} else {
+		record := logDedupRecord{1, time.Now(), time.NewTimer(bufferLength)}
+		a.logDeduplication.logMap[logMsg] = &record
+	}
+	return exists
+}
+
+// logDedupPacket logs information in ob based on disposition and duplication conditions.
+func (a *AntreaPolicyLogger) logDedupPacket(ob *logInfo, bufferLength time.Duration) {
+	// Deduplicate non-Allow packet log.
+	logMsg := fmt.Sprintf("%s %s %s %s SRC: %s DEST: %s %d %s", ob.tableName, ob.npRef, ob.disposition, ob.ofPriority, ob.srcIP, ob.destIP, ob.pktLength, ob.protocolStr)
+	if ob.disposition == openflow.DispositionToString[openflow.DispositionAllow] {
+		a.anpLogger.Printf(logMsg)
+	} else {
+		// Increase count if duplicated within 1 sec, create buffer otherwise.
+		exists := a.updateLogKey(logMsg, bufferLength)
+		if !exists {
+			// Go routine for logging when buffer timer stops.
+			go a.logAfterTimer(logMsg)
+		}
+	}
+}
+
+// newAntreaPolicyLogger is called while newing Antrea network policy agent controller.
+// Customize AntreaPolicyLogger specifically for Antrea Policies audit logging.
+func newAntreaPolicyLogger() (*AntreaPolicyLogger, error) {
+	logDir := filepath.Join(logdir.GetLogDir(), logfileSubdir)
+	logFile := filepath.Join(logDir, logfileName)
+	_, err := os.Stat(logDir)
+	if os.IsNotExist(err) {
+		os.Mkdir(logDir, 0755)
+	} else if err != nil {
+		return nil, fmt.Errorf("received error while accessing Antrea network policy log directory: %v", err)
+	}
+
+	// Use lumberjack log file rotation.
+	logOutput := &lumberjack.Logger{
+		Filename:   logFile,
+		MaxSize:    500,  // allow max 500 megabytes for one log file
+		MaxBackups: 3,    // allow max 3 old log file backups
+		MaxAge:     28,   // allow max 28 days maintenance of old log files
+		Compress:   true, // compress the old log files for backup
+	}
+
+	cAntreaPolicyLogger := &AntreaPolicyLogger{
+		anpLogger:        log.New(logOutput, "", log.Ldate|log.Lmicroseconds),
+		logDeduplication: logRecordDedupMap{logMap: make(map[string]*logDedupRecord)},
+	}
+	klog.V(2).Infof("Initialized Antrea-native Policy Logger for audit logging with log file '%s'", logFile)
+	return cAntreaPolicyLogger, nil
+}
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
index d29cd7069..ded5de464 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
@@ -65,8 +65,6 @@ type Controller struct {
 	antreaPolicyEnabled bool
 	// statusManagerEnabled indicates whether a statusManager is configured.
 	statusManagerEnabled bool
-	// loggingEnabled indicates where Antrea policy audit logging is enabled.
-	loggingEnabled bool
 	// antreaClientProvider provides interfaces to get antreaClient, which can be
 	// used to watch Antrea AddressGroups, AppliedToGroups, and NetworkPolicies.
 	// We need to get antreaClient dynamically because the apiserver cert can be
@@ -83,7 +81,8 @@ type Controller struct {
 	// NetworkPolicy rules with the actual state of Openflow entries.
 	reconciler Reconciler
 	// ofClient registers packetin for Antrea Policy logging.
-	ofClient openflow.Client
+	ofClient           openflow.Client
+	antreaPolicyLogger *AntreaPolicyLogger
 	// statusManager syncs NetworkPolicy statuses with the antrea-controller.
 	// It's only for Antrea NetworkPolicies.
 	statusManager         StatusManager
@@ -104,7 +103,6 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 	entityUpdates <-chan types.EntityReference,
 	antreaPolicyEnabled bool,
 	statusManagerEnabled bool,
-	loggingEnabled bool,
 	denyConnStore *connections.DenyConnectionStore,
 	asyncRuleDeleteInterval time.Duration) (*Controller, error) {
 	c := &Controller{
@@ -114,7 +112,6 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 		ofClient:             ofClient,
 		antreaPolicyEnabled:  antreaPolicyEnabled,
 		statusManagerEnabled: statusManagerEnabled,
-		loggingEnabled:       loggingEnabled,
 		denyConnStore:        denyConnStore,
 	}
 	c.ruleCache = newRuleCache(c.enqueueRule, entityUpdates)
@@ -128,14 +125,15 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 	// Wait until appliedToGroupWatcher, addressGroupWatcher and networkPolicyWatcher to receive bookmark event.
 	c.fullSyncGroup.Add(3)
 
-	if c.ofClient != nil && loggingEnabled {
+	if c.ofClient != nil && antreaPolicyEnabled {
 		// Register packetInHandler
 		c.ofClient.RegisterPacketInHandler(uint8(openflow.PacketInReasonNP), "networkpolicy", c)
 		// Initiate logger for Antrea Policy audit logging
-		err := initLogger()
+		antreaPolicyLogger, err := newAntreaPolicyLogger()
 		if err != nil {
 			return nil, err
 		}
+		c.antreaPolicyLogger = antreaPolicyLogger
 	}
 
 	// Use nodeName to filter resources when watching resources.
diff --git a/pkg/agent/controller/networkpolicy/packetin.go b/pkg/agent/controller/networkpolicy/packetin.go
index 5f531e5d5..5d6da0ece 100644
--- a/pkg/agent/controller/networkpolicy/packetin.go
+++ b/pkg/agent/controller/networkpolicy/packetin.go
@@ -18,16 +18,12 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"log"
-	"os"
-	"path/filepath"
 	"time"
 
 	"antrea.io/libOpenflow/openflow13"
 	"antrea.io/libOpenflow/protocol"
 	"antrea.io/ofnet/ofctrl"
 	"github.com/vmware/go-ipfix/pkg/registry"
-	"gopkg.in/natefinch/lumberjack.v2"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/config"
@@ -35,13 +31,9 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/util/ip"
-	"antrea.io/antrea/pkg/util/logdir"
 )
 
 const (
-	logfileSubdir string = "networkpolicy"
-	logfileName   string = "np.log"
-
 	IPv4HdrLen uint16 = 20
 	IPv6HdrLen uint16 = 40
 
@@ -57,44 +49,6 @@ const (
 	ICMPv6DstAdminProhibitedCode uint8 = 1
 )
 
-var (
-	AntreaPolicyLogger *log.Logger
-)
-
-// 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
-	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
-	destIP      string // destination IP of the traffic logged
-	pktLength   uint16 // packet length of packetin
-	protocolStr string // protocol of the traffic logged
-}
-
-// initLogger is called while newing Antrea network policy agent controller.
-// Customize AntreaPolicyLogger specifically for Antrea Policies audit logging.
-func initLogger() error {
-	logDir := filepath.Join(logdir.GetLogDir(), logfileSubdir)
-	logFile := filepath.Join(logDir, logfileName)
-	if _, err := os.Stat(logDir); os.IsNotExist(err) {
-		os.Mkdir(logDir, 0755)
-	}
-
-	// Use lumberjack log file rot
-	logOutput := &lumberjack.Logger{
-		Filename:   logFile,
-		MaxSize:    500,  // allow max 500 megabytes for one log file
-		MaxBackups: 3,    // allow max 3 old log file backups
-		MaxAge:     28,   // allow max 28 days maintenance of old log files
-		Compress:   true, // compress the old log files for backup
-	}
-	AntreaPolicyLogger = log.New(logOutput, "", log.Ldate|log.Lmicroseconds)
-	klog.V(2).Infof("Initialized Antrea-native Policy Logger for audit logging with log file '%s'", logFile)
-	return nil
-}
-
 // HandlePacketIn is the packetin handler registered to openflow by Antrea network
 // policy agent controller. It performs the appropriate operations based on which
 // bits are set in the "custom reasons" field of the packet received from OVS.
@@ -130,28 +84,6 @@ func (c *Controller) HandlePacketIn(pktIn *ofctrl.PacketIn) error {
 	return nil
 }
 
-// logPacket retrieves information from openflow reg, controller cache, packet-in
-// packet to log.
-func (c *Controller) logPacket(pktIn *ofctrl.PacketIn) error {
-	ob := new(logInfo)
-
-	// Get Network Policy log info
-	err := getNetworkPolicyInfo(pktIn, c, ob)
-	if err != nil {
-		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
-	}
-
-	// Get packet log info
-	err = getPacketInfo(pktIn, ob)
-	if err != nil {
-		return fmt.Errorf("received error while handling packetin for NetworkPolicy: %v", err)
-	}
-
-	// Store log file
-	AntreaPolicyLogger.Printf("%s %s %s %s SRC: %s DEST: %s %d %s", ob.tableName, ob.npRef, ob.disposition, ob.ofPriority, ob.srcIP, ob.destIP, ob.pktLength, ob.protocolStr)
-	return nil
-}
-
 // getMatchRegField returns match to the regNum register.
 func getMatchRegField(matchers *ofctrl.Matchers, regNum uint32) *ofctrl.MatchField {
 	return matchers.GetMatchByName(fmt.Sprintf("NXM_NX_REG%d", regNum))
@@ -242,6 +174,28 @@ func getPacketInfo(pktIn *ofctrl.PacketIn, ob *logInfo) error {
 	return 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.
+func (c *Controller) logPacket(pktIn *ofctrl.PacketIn) error {
+	ob := new(logInfo)
+
+	// Get Network Policy log info
+	err := getNetworkPolicyInfo(pktIn, c, ob)
+	if err != nil {
+		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
+	}
+	// Get packet log info
+	err = getPacketInfo(pktIn, ob)
+	if err != nil {
+		return fmt.Errorf("received error while retrieving NetworkPolicy info: %v", err)
+	}
+
+	// Log the ob info to corresponding file w/ deduplication
+	c.antreaPolicyLogger.logDedupPacket(ob, time.Second)
+	return nil
+}
+
 // rejectRequest sends reject response to the requesting client, based on the
 // packet-in message.
 func (c *Controller) rejectRequest(pktIn *ofctrl.PacketIn) error {
diff --git a/pkg/agent/nodeportlocal/k8s/npl_controller.go b/pkg/agent/nodeportlocal/k8s/npl_controller.go
index a345452c3..6fa0b07a8 100644
--- a/pkg/agent/nodeportlocal/k8s/npl_controller.go
+++ b/pkg/agent/nodeportlocal/k8s/npl_controller.go
@@ -130,12 +130,8 @@ func (c *NPLController) Run(stopCh <-chan struct{}) {
 	if !cache.WaitForNamedCacheSync(controllerName, stopCh, c.podInformer.HasSynced, c.svcInformer.HasSynced) {
 		return
 	}
-	klog.Info("Will fetch Pods and generate NodePortLocal rules for these Pods")
 
-	if err := c.GetPodsAndGenRules(); err != nil {
-		klog.Errorf("Error in getting Pods and generating rules: %v", err)
-		return
-	}
+	c.waitForRulesInitialization()
 
 	for i := 0; i < numWorkers; i++ {
 		go wait.Until(c.Worker, time.Second, stopCh)
@@ -551,18 +547,27 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 	return nil
 }
 
-// GetPodsAndGenRules fetches all the Pods on this Node and looks for valid NodePortLocal
+// waitForRulesInitialization fetches all the Pods on this Node and looks for valid NodePortLocal
 // annotations. If they exist, with a valid Node port, it adds the Node port to the port table and
 // rules. If the NodePortLocal annotation is invalid (cannot be unmarshalled), the annotation is
 // cleared. If the Node port is invalid (maybe the port range was changed and the Agent was
 // restarted), the annotation is ignored and will be removed by the Pod event handlers. The Pod
 // event handlers will also take care of allocating a new Node port if required.
-func (c *NPLController) GetPodsAndGenRules() error {
+// The function is meant to be called during Controller initialization, after the caches have
+// synced. It will block until iptables rules have been synced successfully based on the listed
+// Pods. After it returns, the Controller should start handling events. In case of an unexpected
+// error, the function can return early or may not complete initialization. The Controller's event
+// handlers are able to recover from these errors.
+func (c *NPLController) waitForRulesInitialization() {
+	klog.InfoS("Will fetch Pods and generate NodePortLocal rules for these Pods")
+
 	podList, err := c.podLister.List(labels.Everything())
 	if err != nil {
-		return fmt.Errorf("error in fetching the Pods for Node %s: %v", c.nodeName, err)
+		klog.ErrorS(err, "Error when listing Pods for Node")
 	}
 
+	// in case of an error when listing Pods above, allNPLPorts will be
+	// empty and all NPL iptables rules will be deleted.
 	allNPLPorts := []rules.PodNodePort{}
 	for i := range podList {
 		// For each Pod:
@@ -576,12 +581,11 @@ func (c *NPLController) GetPodsAndGenRules() error {
 			continue
 		}
 		nplData := []NPLAnnotation{}
-		err := json.Unmarshal([]byte(nplAnnotation), &nplData)
-		if err != nil {
+		if err := json.Unmarshal([]byte(nplAnnotation), &nplData); err != nil {
+			klog.InfoS("Found invalid NodePortLocal annotation for Pod that cannot be parsed, cleaning it up", "pod", klog.KObj(pod))
 			// if there's an error in this NodePortLocal annotation, clean it up
-			err := c.cleanupNPLAnnotationForPod(pod)
-			if err != nil {
-				return err
+			if err := c.cleanupNPLAnnotationForPod(pod); err != nil {
+				klog.ErrorS(err, "Error when cleaning up NodePortLocal annotation for Pod", "pod", klog.KObj(pod))
 			}
 			continue
 		}
@@ -590,27 +594,30 @@ func (c *NPLController) GetPodsAndGenRules() error {
 			if npl.NodePort > c.portTable.EndPort || npl.NodePort < c.portTable.StartPort {
 				// ignoring annotation for now, it will be removed by the first call
 				// to handleAddUpdatePod
-				klog.V(2).Infof("Found invalid NodePortLocal annotation for Pod %s/%s: %s, ignoring it", pod.Namespace, pod.Name, nplAnnotation)
+				klog.V(2).InfoS("Found NodePortLocal annotation for which the allocated port doesn't fall into the configured range", "pod", klog.KObj(pod))
 				continue
-			} else {
-				allNPLPorts = append(allNPLPorts, rules.PodNodePort{
-					NodePort: npl.NodePort,
-					PodPort:  npl.PodPort,
-					PodIP:    pod.Status.PodIP,
-				})
 			}
+			allNPLPorts = append(allNPLPorts, rules.PodNodePort{
+				NodePort: npl.NodePort,
+				PodPort:  npl.PodPort,
+				PodIP:    pod.Status.PodIP,
+			})
 		}
 	}
 
-	if err := c.addRulesForNPLPorts(allNPLPorts); err != nil {
-		return err
+	rulesInitialized := make(chan struct{})
+	if err := c.addRulesForNPLPorts(allNPLPorts, rulesInitialized); err != nil {
+		klog.ErrorS(err, "Cannot install NodePortLocal rules")
+		return
 	}
 
-	return nil
+	klog.InfoS("Waiting for initialization of NodePortLocal rules to complete")
+	<-rulesInitialized
+	klog.InfoS("Initialization of NodePortLocal rules successful")
 }
 
-func (c *NPLController) addRulesForNPLPorts(allNPLPorts []rules.PodNodePort) error {
-	return c.portTable.SyncRules(allNPLPorts)
+func (c *NPLController) addRulesForNPLPorts(allNPLPorts []rules.PodNodePort, synced chan<- struct{}) error {
+	return c.portTable.RestoreRules(allNPLPorts, synced)
 }
 
 // cleanupNPLAnnotationForPod removes the NodePortLocal annotation from the Pod's annotations map entirely.
diff --git a/pkg/agent/nodeportlocal/portcache/port_table.go b/pkg/agent/nodeportlocal/portcache/port_table.go
index 19a776913..97d30ff8d 100644
--- a/pkg/agent/nodeportlocal/portcache/port_table.go
+++ b/pkg/agent/nodeportlocal/portcache/port_table.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net"
 	"sync"
+	"time"
 
 	"k8s.io/klog/v2"
 
@@ -156,8 +157,26 @@ func (pt *PortTable) RuleExists(podIP string, podPort int) bool {
 	return false
 }
 
-func (pt *PortTable) SyncRules(allNPLPorts []rules.PodNodePort) error {
-	validNPLPorts := make([]rules.PodNodePort, 0, len(allNPLPorts))
+// syncRules ensures that contents of the port table matches the iptables rules present on the Node.
+func (pt *PortTable) syncRules() error {
+	pt.tableLock.Lock()
+	defer pt.tableLock.Unlock()
+	nplPorts := make([]rules.PodNodePort, 0, len(pt.Table))
+	for _, data := range pt.Table {
+		nplPorts = append(nplPorts, rules.PodNodePort{
+			NodePort: data.NodePort,
+			PodPort:  data.PodPort,
+			PodIP:    data.PodIP,
+		})
+	}
+	return pt.PodPortRules.AddAllRules(nplPorts)
+}
+
+// RestoreRules should be called on startup to restore a set of NPL rules. It is non-blocking but
+// takes as a parameter a channel, synced, which will be closed when the necessary rules have been
+// restored successfully. No other operations should be performed on the PortTable until the channel
+// is closed.
+func (pt *PortTable) RestoreRules(allNPLPorts []rules.PodNodePort, synced chan<- struct{}) error {
 	pt.tableLock.Lock()
 	defer pt.tableLock.Unlock()
 	for _, nplPort := range allNPLPorts {
@@ -176,9 +195,22 @@ func (pt *PortTable) SyncRules(allNPLPorts []rules.PodNodePort) error {
 			socket:   socket,
 		}
 		pt.Table[nplPort.NodePort] = data
-		validNPLPorts = append(validNPLPorts, nplPort)
 	}
-	return pt.PodPortRules.AddAllRules(validNPLPorts)
+	// retry mechanism as iptables-restore can fail if other components (in Antrea or other
+	// software) are accessing iptables.
+	go func() {
+		defer close(synced)
+		var backoffTime = 2 * time.Second
+		for {
+			if err := pt.syncRules(); err != nil {
+				klog.ErrorS(err, "Failed to restore iptables rules", "backoff", backoffTime)
+				time.Sleep(backoffTime)
+				continue
+			}
+			break
+		}
+	}()
+	return nil
 }
 
 // openLocalPort binds to the provided port.
diff --git a/pkg/ovs/ovsctl/appctl.go b/pkg/ovs/ovsctl/appctl.go
index 024e6c47f..cf1e36eb6 100644
--- a/pkg/ovs/ovsctl/appctl.go
+++ b/pkg/ovs/ovsctl/appctl.go
@@ -15,10 +15,13 @@
 package ovsctl
 
 import (
+	"bufio"
 	"bytes"
 	"fmt"
 	"net"
 	"strings"
+
+	"k8s.io/klog/v2"
 )
 
 // Shell exits with 127 if the command to execute is not found.
@@ -33,12 +36,34 @@ var (
 	nonIPDLTypes = []string{"arp", "rarp", "dl_type="}
 )
 
+type DPFeature string
+
+const (
+	CTStateFeature    DPFeature = "CT state"
+	CTZoneFeature     DPFeature = "CT zone"
+	CTMarkFeature     DPFeature = "CT mark"
+	CTLabelFeature    DPFeature = "CT label"
+	CTStateNATFeature DPFeature = "CT state NAT"
+)
+
+var knownFeatures = map[DPFeature]struct{}{
+	CTStateFeature:    {},
+	CTZoneFeature:     {},
+	CTMarkFeature:     {},
+	CTLabelFeature:    {},
+	CTStateNATFeature: {},
+}
+
 type ovsCtlClient struct {
 	bridge string
+	// To allow injection for testing.
+	runAppCtl func(cmd string, needsBridge bool, args ...string) ([]byte, *ExecError)
 }
 
 func NewClient(bridge string) *ovsCtlClient {
-	return &ovsCtlClient{bridge}
+	client := &ovsCtlClient{bridge: bridge}
+	client.runAppCtl = client.RunAppctlCmd
+	return client
 }
 
 func newBadRequestError(msg string) BadRequestError {
@@ -138,7 +163,7 @@ func getNwDstKey(ip net.IP) (string, string) {
 }
 
 func (c *ovsCtlClient) runTracing(flow string) (string, error) {
-	out, execErr := c.RunAppctlCmd("ofproto/trace", true, flow)
+	out, execErr := c.runAppCtl("ofproto/trace", true, flow)
 	if execErr != nil {
 		return "", execErr
 	}
@@ -166,3 +191,39 @@ func (c *ovsCtlClient) RunAppctlCmd(cmd string, needsBridge bool, args ...string
 	}
 	return out, nil
 }
+
+func (c *ovsCtlClient) GetDPFeatures() (map[DPFeature]bool, error) {
+	cmd := "dpif/show-dp-features"
+	out, err := c.runAppCtl(cmd, true)
+	if err != nil {
+		return nil, fmt.Errorf("error listing DP features: %v", err)
+	}
+	scanner := bufio.NewScanner(strings.NewReader(string(out)))
+	scanner.Split(bufio.ScanLines)
+	features := map[DPFeature]bool{}
+	for scanner.Scan() {
+		line := scanner.Text()
+		fields := strings.Split(line, ":")
+		if len(fields) != 2 {
+			klog.InfoS("Unexpected output from dpif/show-dp-features", "line", line)
+			continue
+		}
+		feature := DPFeature(strings.TrimSpace(fields[0]))
+		_, known := knownFeatures[feature]
+		if !known {
+			continue
+		}
+		value := strings.TrimSpace(fields[1])
+		var supported bool
+		if value == "Yes" {
+			supported = true
+		} else if value == "No" {
+			supported = false
+		} else {
+			klog.InfoS("Unexpected non boolean value", "feature", feature, "value", value)
+			continue
+		}
+		features[feature] = supported
+	}
+	return features, nil
+}
diff --git a/pkg/ovs/ovsctl/interface.go b/pkg/ovs/ovsctl/interface.go
index 590624f81..85abf16af 100644
--- a/pkg/ovs/ovsctl/interface.go
+++ b/pkg/ovs/ovsctl/interface.go
@@ -54,6 +54,8 @@ type OVSCtlClient interface {
 	// RunAppctlCmd executes "ovs-appctl" command and returns the outputs.
 	// Some commands are bridge specific and some are not. Passing a bool to distinguish that.
 	RunAppctlCmd(cmd string, needsBridge bool, args ...string) ([]byte, *ExecError)
+	// GetDPFeatures executes "ovs-appctl dpif/show-dp-features" to check supported DP features.
+	GetDPFeatures() (map[DPFeature]bool, error)
 }
 
 type BadRequestError string


Test patch:
diff --git a/ci/jenkins/test.sh b/ci/jenkins/test.sh
index 3abfd066e..18612de7d 100755
--- a/ci/jenkins/test.sh
+++ b/ci/jenkins/test.sh
@@ -138,31 +138,59 @@ function clean_for_windows_install_cni {
 }
 
 function collect_windows_network_info_and_logs {
-    echo "=== Collecting network info after failure ==="
-    mkdir network_info
+    echo "=== Collecting information after failure ==="
+    DEBUG_LOG_PATH="debug_logs"
+    mkdir "${DEBUG_LOG_PATH}"
     kubectl get pod -n kube-system -l component=antrea-agent --no-headers=true | awk '{print $1}' | while read AGENTNAME; do
-        mkdir network_info/${AGENTNAME}
-        kubectl exec "$AGENTNAME" -c antrea-agent -n kube-system -- ovs-ofctl dump-flows br-int > "network_info/${AGENTNAME}/flows" || true
-        kubectl exec "$AGENTNAME" -c antrea-agent -n kube-system -- ovs-vsctl show > "network_info/$AGENTNAME/db" || true
-        kubectl exec "$AGENTNAME" -c antrea-agent -n kube-system -- ip a > "network_info/${AGENTNAME}/addrs" || true
-    done
-    kubectl get nodes -o wide --no-headers=true | awk -v role="$CONTROL_PLANE_NODE_ROLE" '$3 != role && $1 ~ /win/ {print $6}' | while read IP; do
-        mkdir network_info/${IP}
-        ssh -o StrictHostKeyChecking=no -n Administrator@${IP} "powershell.exe get-NetAdapter" > "network_info/${IP}/adapters" || true
-        ssh -o StrictHostKeyChecking=no -n Administrator@${IP} "ipconfig.exe" > "network_info/${IP}/ipconfig" || true
-        ssh -o StrictHostKeyChecking=no -n Administrator@${IP} "powershell.exe Get-HNSNetwork" > "network_info/${IP}/hns_network" || true
-        ssh -o StrictHostKeyChecking=no -n Administrator@${IP} "powershell.exe Get-HNSEndpoint" > "network_info/${IP}/hns_endpoint" || true
+        IP=$(kubectl get pod ${AGENTNAME} -n kube-system -o json | jq -r '.status.hostIP')
+        mkdir "${DEBUG_LOG_PATH}/${AGENTNAME}"
+
+        echo "=== Collecting '${AGENTNAME}' agent network info after failure ==="
+        AGENT_NETWORK_INFO_PATH="${DEBUG_LOG_PATH}/${AGENTNAME}/network_info"
+        mkdir "${AGENT_NETWORK_INFO_PATH}"
+        if [[ "${AGENTNAME}" =~ "windows" ]]; then
+            ssh -o StrictHostKeyChecking=no -n administrator@${IP} "ovs-ofctl dump-flows br-int" > "${AGENT_NETWORK_INFO_PATH}/flows" || true
+            ssh -o StrictHostKeyChecking=no -n administrator@${IP} "ovs-vsctl show" > "${AGENT_NETWORK_INFO_PATH}/db" || true
+            ssh -o StrictHostKeyChecking=no -n administrator@${IP} "ipconfig" > "${AGENT_NETWORK_INFO_PATH}/addrs" || true
+        else
+            kubectl exec "${AGENTNAME}" -c antrea-agent -n kube-system -- ovs-ofctl dump-flows br-int > "${AGENT_NETWORK_INFO_PATH}/flows" || true
+            kubectl exec "${AGENTNAME}" -c antrea-agent -n kube-system -- ovs-vsctl show > "${AGENT_NETWORK_INFO_PATH}/db" || true
+            kubectl exec "${AGENTNAME}" -c antrea-agent -n kube-system -- ip a > "${AGENT_NETWORK_INFO_PATH}/addrs" || true
+        fi
+
+        echo "=== Collecting '${AGENTNAME}' antrea agent log after failure ==="
+        ANTREA_AGENT_LOG_PATH="${DEBUG_LOG_PATH}/${AGENTNAME}/antrea_agent_log"
+        mkdir "${ANTREA_AGENT_LOG_PATH}"
+        kubectl logs "${AGENTNAME}" -n kube-system -c antrea-agent > "${ANTREA_AGENT_LOG_PATH}/antrea-agent.log" || true
+        if [[ "${AGENTNAME}" =~ "windows" ]]; then
+            echo "Windows agent doesn't have antrea-ovs container"
+        else
+            kubectl logs "${AGENTNAME}" -n kube-system -c antrea-ovs > "${ANTREA_AGENT_LOG_PATH}/antrea-ovs.log" || true
+        fi
     done
-    tar zcf network_info.tar.gz network_info
 
-    echo "=== Collecting antrea agent log after failure ==="
-    mkdir antrea_agent_log
-    kubectl get pod -n kube-system -l component=antrea-agent --no-headers=true | awk '{print $1}' | while read AGENTNAME; do
-        mkdir antrea_agent_log/$AGENTNAME
-        kubectl logs $AGENTNAME -n kube-system -c antrea-agent > antrea_agent_log/$AGENTNAME/antrea-agent.log || true
-        kubectl logs $AGENTNAME -n kube-system -c antrea-ovs > antrea_agent_log/$AGENTNAME/antrea-ovs.log || true
+    kubectl get nodes -o wide --no-headers=true | awk -v role="$CONTROL_PLANE_NODE_ROLE" '$3 != role && $1 ~ /win/ {print $1}' | while read NODENAME; do
+        IP=$(kubectl get node ${NODENAME} -o json | jq -r '.status.addresses[] | select(.type | test("InternalIP")).address')
+        mkdir "${DEBUG_LOG_PATH}/${NODENAME}"
+
+        echo "=== Collecting '${NODENAME}' Node network info after failure ==="
+        NODE_NETWORK_INFO_PATH="${DEBUG_LOG_PATH}/${NODENAME}/network_info"
+        mkdir "${NODE_NETWORK_INFO_PATH}"
+        ssh -o StrictHostKeyChecking=no -n administrator@${IP} "powershell.exe get-NetAdapter" > "${NODE_NETWORK_INFO_PATH}/adapters" || true
+        ssh -o StrictHostKeyChecking=no -n administrator@${IP} "ipconfig.exe" > "${NODE_NETWORK_INFO_PATH}/ipconfig" || true
+        ssh -o StrictHostKeyChecking=no -n administrator@${IP} "powershell.exe Get-HNSNetwork" > "${NODE_NETWORK_INFO_PATH}/hns_network" || true
+        ssh -o StrictHostKeyChecking=no -n administrator@${IP} "powershell.exe Get-HNSEndpoint" > "${NODE_NETWORK_INFO_PATH}/hns_endpoint" || true
+
+        echo "=== Collecting '${NODENAME}' kubelet and docker logs after failure ==="
+        KUBELET_LOG_PATH="${DEBUG_LOG_PATH}/${NODENAME}/kubelet"
+        mkdir "${KUBELET_LOG_PATH}"
+        scp -q -o StrictHostKeyChecking=no -T administrator@${IP}:/cygdrive/c/var/log/kubelet/* "${KUBELET_LOG_PATH}"
+
+        DOCKER_LOG_PATH="${DEBUG_LOG_PATH}/${NODENAME}/docker"
+        mkdir "${DOCKER_LOG_PATH}"
+        scp -q -o StrictHostKeyChecking=no -T administrator@${IP}:'/cygdrive/c/"Program Files"/Docker/dockerd.log*' "${DOCKER_LOG_PATH}"
     done
-    tar zcf antrea_agent_log.tar.gz antrea_agent_log
+    tar zcf debug_logs.tar.gz "${DEBUG_LOG_PATH}"
 }
 
 function wait_for_antrea_windows_pods_ready {
diff --git a/pkg/agent/controller/networkpolicy/audit_logging_test.go b/pkg/agent/controller/networkpolicy/audit_logging_test.go
new file mode 100644
index 000000000..80df57219
--- /dev/null
+++ b/pkg/agent/controller/networkpolicy/audit_logging_test.go
@@ -0,0 +1,121 @@
+// Copyright 2021 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 networkpolicy
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// mockLogger implements io.Writer.
+type mockLogger struct {
+	mu     sync.Mutex
+	logged chan string
+}
+
+func (l *mockLogger) Write(p []byte) (n int, err error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if p == nil {
+		return 0, errors.New("error writing to mock Logger")
+	}
+	msg := string(p[:])
+	l.logged <- msg
+	return len(msg), nil
+}
+
+func newTestAntreaPolicyLogger() (*AntreaPolicyLogger, *mockLogger) {
+	mockAnpLogger := &mockLogger{logged: make(chan string, 100)}
+	antreaLogger := &AntreaPolicyLogger{
+		anpLogger:        log.New(mockAnpLogger, "", log.Ldate),
+		logDeduplication: logRecordDedupMap{logMap: make(map[string]*logDedupRecord)},
+	}
+	return antreaLogger, mockAnpLogger
+}
+
+func newLogInfo(disposition string) (*logInfo, string) {
+	expected := "AntreaPolicyIngressRule AntreaNetworkPolicy:default/test " + disposition + " 0 SRC: 0.0.0.0 DEST: 1.1.1.1 60 TCP"
+	return &logInfo{
+		tableName:   "AntreaPolicyIngressRule",
+		npRef:       "AntreaNetworkPolicy:default/test",
+		ofPriority:  "0",
+		disposition: disposition,
+		srcIP:       "0.0.0.0",
+		destIP:      "1.1.1.1",
+		pktLength:   60,
+		protocolStr: "TCP",
+	}, expected
+}
+
+func sendMultiplePackets(antreaLogger *AntreaPolicyLogger, ob *logInfo, numPackets int) {
+	count := 0
+	for range time.Tick(12 * time.Millisecond) {
+		count += 1
+		antreaLogger.logDedupPacket(ob, 100*time.Millisecond)
+		if count == numPackets {
+			break
+		}
+	}
+}
+
+func expectedLogWithCount(msg string, count int) string {
+	return fmt.Sprintf("%s [%d", msg, count)
+}
+
+func TestAllowPacketLog(t *testing.T) {
+	antreaLogger, mockAnpLogger := newTestAntreaPolicyLogger()
+	ob, expected := newLogInfo("Allow")
+
+	antreaLogger.logDedupPacket(ob, 100*time.Millisecond)
+	actual := <-mockAnpLogger.logged
+	assert.Contains(t, actual, expected)
+}
+
+func TestDropPacketLog(t *testing.T) {
+	antreaLogger, mockAnpLogger := newTestAntreaPolicyLogger()
+	ob, expected := newLogInfo("Drop")
+
+	antreaLogger.logDedupPacket(ob, 100*time.Millisecond)
+	actual := <-mockAnpLogger.logged
+	assert.Contains(t, actual, expected)
+}
+
+func TestDropPacketDedupLog(t *testing.T) {
+	antreaLogger, mockAnpLogger := newTestAntreaPolicyLogger()
+	ob, expected := newLogInfo("Drop")
+	// add the additional log info for duplicate packets
+	expected = expectedLogWithCount(expected, 2)
+
+	go sendMultiplePackets(antreaLogger, ob, 2)
+	actual := <-mockAnpLogger.logged
+	assert.Contains(t, actual, expected)
+}
+
+func TestDropPacketMultiDedupLog(t *testing.T) {
+	antreaLogger, mockAnpLogger := newTestAntreaPolicyLogger()
+	ob, expected := newLogInfo("Drop")
+
+	go sendMultiplePackets(antreaLogger, ob, 10)
+	actual := <-mockAnpLogger.logged
+	assert.Contains(t, actual, expectedLogWithCount(expected, 9))
+	actual = <-mockAnpLogger.logged
+	assert.Contains(t, actual, expected)
+}
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
index b45ed0270..0547ec13a 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
@@ -53,9 +53,10 @@ func newTestController() (*Controller, *fake.Clientset, *mockReconciler) {
 	clientset := &fake.Clientset{}
 	ch := make(chan agenttypes.EntityReference, 100)
 	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", ch,
-		true, true, true, nil, testAsyncDeleteInterval)
+		true, true, nil, testAsyncDeleteInterval)
 	reconciler := newMockReconciler()
 	controller.reconciler = reconciler
+	controller.antreaPolicyLogger = nil
 	return controller, clientset, reconciler
 }
 
diff --git a/pkg/agent/controller/networkpolicy/packetin_test.go b/pkg/agent/controller/networkpolicy/packetin_test.go
index f92c57ac4..c7cd3f684 100644
--- a/pkg/agent/controller/networkpolicy/packetin_test.go
+++ b/pkg/agent/controller/networkpolicy/packetin_test.go
@@ -25,10 +25,6 @@ import (
 )
 
 func TestGetPacketInfo(t *testing.T) {
-	type args struct {
-		pktIn *ofctrl.PacketIn
-		ob    *logInfo
-	}
 	tests := []struct {
 		name       string
 		pktIn      *ofctrl.PacketIn
diff --git a/pkg/agent/nodeportlocal/npl_agent_test.go b/pkg/agent/nodeportlocal/npl_agent_test.go
index 2a77df15c..e8fa2994c 100644
--- a/pkg/agent/nodeportlocal/npl_agent_test.go
+++ b/pkg/agent/nodeportlocal/npl_agent_test.go
@@ -154,12 +154,11 @@ func getTestSvcWithPortName(portName string) *corev1.Service {
 
 type testData struct {
 	*testing.T
-	stopCh         chan struct{}
-	ctrl           *gomock.Controller
-	k8sClient      *k8sfake.Clientset
-	portTable      *portcache.PortTable
-	mockPortOpener *portcachetesting.MockLocalPortOpener
-	wg             sync.WaitGroup
+	stopCh    chan struct{}
+	ctrl      *gomock.Controller
+	k8sClient *k8sfake.Clientset
+	portTable *portcache.PortTable
+	wg        sync.WaitGroup
 }
 
 func (data *testData) runWrapper(c *nplk8s.NPLController) {
@@ -170,18 +169,25 @@ func (data *testData) runWrapper(c *nplk8s.NPLController) {
 	}()
 }
 
+type customizePortOpenerExpectations func(*portcachetesting.MockLocalPortOpener)
+type customizePodPortRulesExpectations func(*rulestesting.MockPodPortRules)
+
 type testConfig struct {
-	defaultPortOpenerExpectations bool
+	customPortOpenerExpectations   customizePortOpenerExpectations
+	customPodPortRulesExpectations customizePodPortRulesExpectations
 }
 
 func newTestConfig() *testConfig {
-	return &testConfig{
-		defaultPortOpenerExpectations: true,
-	}
+	return &testConfig{}
+}
+
+func (tc *testConfig) withCustomPortOpenerExpectations(fn customizePortOpenerExpectations) *testConfig {
+	tc.customPortOpenerExpectations = fn
+	return tc
 }
 
-func (tc *testConfig) withDefaultPortOpenerExpectations(v bool) *testConfig {
-	tc.defaultPortOpenerExpectations = false
+func (tc *testConfig) withCustomPodPortRulesExpectations(fn customizePodPortRulesExpectations) *testConfig {
+	tc.customPodPortRulesExpectations = fn
 	return tc
 }
 
@@ -191,22 +197,27 @@ func setUp(t *testing.T, tc *testConfig, objects ...runtime.Object) *testData {
 	mockCtrl := gomock.NewController(t)
 
 	mockIPTables := rulestesting.NewMockPodPortRules(mockCtrl)
-	mockIPTables.EXPECT().AddRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
-	mockIPTables.EXPECT().DeleteRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
-	mockIPTables.EXPECT().AddAllRules(gomock.Any()).AnyTimes()
+	if tc.customPodPortRulesExpectations != nil {
+		tc.customPodPortRulesExpectations(mockIPTables)
+	} else {
+		mockIPTables.EXPECT().AddRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+		mockIPTables.EXPECT().DeleteRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+		mockIPTables.EXPECT().AddAllRules(gomock.Any()).AnyTimes()
+	}
 
 	mockPortOpener := portcachetesting.NewMockLocalPortOpener(mockCtrl)
-	if tc.defaultPortOpenerExpectations {
+	if tc.customPortOpenerExpectations != nil {
+		tc.customPortOpenerExpectations(mockPortOpener)
+	} else {
 		mockPortOpener.EXPECT().OpenLocalPort(gomock.Any()).AnyTimes().Return(&fakeSocket{}, nil)
 	}
 
 	data := &testData{
-		T:              t,
-		stopCh:         make(chan struct{}),
-		ctrl:           mockCtrl,
-		k8sClient:      k8sfake.NewSimpleClientset(objects...),
-		portTable:      newPortTable(mockIPTables, mockPortOpener),
-		mockPortOpener: mockPortOpener,
+		T:         t,
+		stopCh:    make(chan struct{}),
+		ctrl:      mockCtrl,
+		k8sClient: k8sfake.NewSimpleClientset(objects...),
+		portTable: newPortTable(mockIPTables, mockPortOpener),
 	}
 
 	// informerFactory is initialized and started from cmd/antrea-agent/agent.go
@@ -658,30 +669,35 @@ var (
 // TestNodePortAlreadyBoundTo validates that when a port is already bound to, a different port will
 // be selected for NPL.
 func TestNodePortAlreadyBoundTo(t *testing.T) {
-	testSvc := getTestSvc()
-	testPod := getTestPod()
-	testConfig := newTestConfig().withDefaultPortOpenerExpectations(false)
-	testData := setUp(t, testConfig)
-	defer testData.tearDown()
-
 	var nodePort int
-	gomock.InOrder(
-		testData.mockPortOpener.EXPECT().OpenLocalPort(gomock.Any()).Return(nil, portTakenError),
-		testData.mockPortOpener.EXPECT().OpenLocalPort(gomock.Any()).DoAndReturn(func(port int) (portcache.Closeable, error) {
-			nodePort = port
-			return &fakeSocket{}, nil
-		}),
-	)
-
-	_, err := testData.k8sClient.CoreV1().Services(defaultNS).Create(context.TODO(), testSvc, metav1.CreateOptions{})
-	require.NoError(t, err, "Service creation failed")
-
-	_, err = testData.k8sClient.CoreV1().Pods(defaultNS).Create(context.TODO(), testPod, metav1.CreateOptions{})
-	require.NoError(t, err, "Pod creation failed")
+	testConfig := newTestConfig().withCustomPortOpenerExpectations(func(mockPortOpener *portcachetesting.MockLocalPortOpener) {
+		gomock.InOrder(
+			mockPortOpener.EXPECT().OpenLocalPort(gomock.Any()).Return(nil, portTakenError),
+			mockPortOpener.EXPECT().OpenLocalPort(gomock.Any()).DoAndReturn(func(port int) (portcache.Closeable, error) {
+				nodePort = port
+				return &fakeSocket{}, nil
+			}),
+		)
+	})
+	testData, _, testPod := setUpWithTestServiceAndPod(t, testConfig)
+	defer testData.tearDown()
 
 	value, err := testData.pollForPodAnnotation(testPod.Name, true)
 	require.NoError(t, err, "Poll for annotation check failed")
 	annotation := testData.checkAnnotationValue(value, defaultPort)[0] // length of slice is guaranteed to be correct at this stage
 	assert.Equal(t, nodePort, annotation.NodePort)
-	assert.True(t, testData.portTable.RuleExists(defaultPodIP, defaultPort))
+}
+
+func TestSyncRulesError(t *testing.T) {
+	testConfig := newTestConfig().withCustomPodPortRulesExpectations(func(mockIPTables *rulestesting.MockPodPortRules) {
+		mockIPTables.EXPECT().AddRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+		mockIPTables.EXPECT().DeleteRule(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
+		gomock.InOrder(
+			mockIPTables.EXPECT().AddAllRules(gomock.Any()).Return(fmt.Errorf("iptables failure")),
+			mockIPTables.EXPECT().AddAllRules(gomock.Any()).Return(nil).AnyTimes(),
+		)
+	})
+
+	testData, _, _ := setUpWithTestServiceAndPod(t, testConfig)
+	defer testData.tearDown()
 }
diff --git a/pkg/ovs/ovsctl/appctl_test.go b/pkg/ovs/ovsctl/appctl_test.go
new file mode 100644
index 000000000..5b3ae57f8
--- /dev/null
+++ b/pkg/ovs/ovsctl/appctl_test.go
@@ -0,0 +1,112 @@
+// Copyright 2021 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 ovsctl
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOvsCtlClientGetDPFeatures(t *testing.T) {
+	tests := []struct {
+		name   string
+		output string
+		want   map[DPFeature]bool
+	}{
+		{
+			name: "fully supported",
+			output: `Masked set action: Yes
+Tunnel push pop: No
+Ufid: Yes
+Truncate action: Yes
+Clone action: No
+Sample nesting: 10
+Conntrack eventmask: Yes
+Conntrack clear: Yes
+Max dp_hash algorithm: 0
+Check pkt length action: No
+Conntrack timeout policy: No
+Explicit Drop action: No
+Optimized Balance TCP mode: No
+Max VLAN headers: 2
+Max MPLS depth: 1
+Recirc: Yes
+CT state: Yes
+CT zone: Yes
+CT mark: Yes
+CT label: Yes
+CT state NAT: Yes
+CT orig tuple: Yes
+CT orig tuple for IPv6: Yes
+IPv6 ND Extension: No`,
+			want: map[DPFeature]bool{
+				CTStateFeature:    true,
+				CTZoneFeature:     true,
+				CTMarkFeature:     true,
+				CTLabelFeature:    true,
+				CTStateNATFeature: true,
+			},
+		},
+		{
+			name: "partially supported",
+			output: `Masked set action: Yes
+Tunnel push pop: No
+Ufid: Yes
+Truncate action: No
+Clone action: No
+Sample nesting: 3
+Conntrack eventmask: No
+Conntrack clear: No
+Max dp_hash algorithm: 0
+Check pkt length action: No
+Conntrack timeout policy: No
+Explicit Drop action: No
+Optimized Balance TCP mode: No
+Max VLAN headers: 1
+Max MPLS depth: 1
+Recirc: Yes
+CT state: Yes
+CT zone: Yes
+CT mark: Yes
+CT label: Yes
+CT state NAT: No
+CT orig tuple: No
+CT orig tuple for IPv6: No
+IPv6 ND Extension: No`,
+			want: map[DPFeature]bool{
+				CTStateFeature:    true,
+				CTZoneFeature:     true,
+				CTMarkFeature:     true,
+				CTLabelFeature:    true,
+				CTStateNATFeature: false,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &ovsCtlClient{
+				bridge: "br-int",
+				runAppCtl: func(cmd string, needsBridge bool, args ...string) ([]byte, *ExecError) {
+					return []byte(tt.output), nil
+				},
+			}
+			got, err := c.GetDPFeatures()
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
diff --git a/pkg/ovs/ovsctl/testing/mock_ovsctl.go b/pkg/ovs/ovsctl/testing/mock_ovsctl.go
index a186896c2..7b1f3cc7e 100644
--- a/pkg/ovs/ovsctl/testing/mock_ovsctl.go
+++ b/pkg/ovs/ovsctl/testing/mock_ovsctl.go
@@ -142,6 +142,21 @@ func (mr *MockOVSCtlClientMockRecorder) DumpTableFlows(arg0 interface{}) *gomock
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DumpTableFlows", reflect.TypeOf((*MockOVSCtlClient)(nil).DumpTableFlows), arg0)
 }
 
+// GetDPFeatures mocks base method
+func (m *MockOVSCtlClient) GetDPFeatures() (map[ovsctl.DPFeature]bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetDPFeatures")
+	ret0, _ := ret[0].(map[ovsctl.DPFeature]bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetDPFeatures indicates an expected call of GetDPFeatures
+func (mr *MockOVSCtlClientMockRecorder) GetDPFeatures() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDPFeatures", reflect.TypeOf((*MockOVSCtlClient)(nil).GetDPFeatures))
+}
+
 // RunAppctlCmd mocks base method
 func (m *MockOVSCtlClient) RunAppctlCmd(arg0 string, arg1 bool, arg2 ...string) ([]byte, *ovsctl.ExecError) {
 	m.ctrl.T.Helper()