Repository: antrea-io/antrea (go)

Issue:
antrea-agent Pods didn't update in rolling update fashion in large scale cluster
**Describe the bug**
<!--
A clear and concise description of what the bug is.
If you believe this bug is a security issue, please don't use this template and follow our [security guidelines](/SECURITY.md)
-->
Thanks @alokmaurya88 for reporting this issue.
In a large scale cluster, we saw that when antrea is getting upgraded, most of the agents go in terminating state at once, just after controller is upgraded, even though the updateStrategy was set to RollingUpdate.
```
# k1 rollout status daemonset antrea-agent -n kube-system
Waiting for daemon set "antrea-agent" rollout to finish: 0 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 0 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 1 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 27 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 106 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 122 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 127 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 128 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 129 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 130 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 130 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 131 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 132 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 133 out of 503 new pods have been updated...
Waiting for daemon set "antrea-agent" rollout to finish: 133 out of 503 new pods have been updated..
```

This was caused by NotReady state of antrea-agent Pods. Since v1.0, antrea-agent's readiness probe added its connection with antrea-controller as a check condition. It seems when antrea-controller restarted, many agent Pods cannot recover their connection with antrea-controller in 40 seconds, then they were reported as NotReady, and DaemonSet controller removed these Pods immediately.

The reason why some agents cannot recover their connection with antrea-controller in 40 seconds is because it indeed needs that time in such scale. Take one antrea-agent's log and antrea-controller's log as example:
1. antrea-agent got disconnected from antrea-controller at 12:10:48 because antrea-controller was updated and terminated
```
I0804 12:10:48.589814       1 networkpolicy_controller.go:632] Starting watch for AddressGroup
I0804 12:10:48.589371       1 networkpolicy_controller.go:632] Starting watch for AppliedToGroup
I0804 12:10:48.590033       1 networkpolicy_controller.go:632] Starting watch for NetworkPolicy
W0804 12:10:48.627768       1 networkpolicy_controller.go:635] Failed to start watch for NetworkPolicy: Get "https://10.8.36.180:443/apis/controlplane.antrea.io/v1beta2/networkpolicies?fieldSelector=nodeName%3Dtkg-scale-workload-md-0-d8b7b4cb4-hztxq&watch=true": dial tcp 10.8.36.180:443: connect: connection refused
W0804 12:10:48.627656       1 networkpolicy_controller.go:635] Failed to start watch for AppliedToGroup: Get "https://10.8.36.180:443/apis/controlplane.antrea.io/v1beta2/appliedtogroups?fieldSelector=nodeName%3Dtkg-scale-workload-md-0-d8b7b4cb4-hztxq&watch=true": dial tcp 10.8.36.180:443: connect: connection refused
W0804 12:10:48.627318       1 networkpolicy_controller.go:635] Failed to start watch for AddressGroup: Get "https://10.8.36.180:443/apis/controlplane.antrea.io/v1beta2/addressgroups?fieldSelector=nodeName%3Dtkg-scale-workload-md-0-d8b7b4cb4-hztxq&watch=true": dial tcp 10.8.36.180:443: connect: connection refused
```
2. new antrea-controller started at 12:11:02, so it took 14s for Kubernetes to destory and create the Pod.
```
I0804 12:11:02.090475       1 log_file.go:99] Set log file max size to 104857600
I0804 12:11:02.091361       1 feature_gate.go:243] feature gates: &{map[]}
I0804 12:11:02.091396       1 controller.go:100] Starting Antrea Controller (version v1.2.0-1079731)
```
3. antrea-controller was not ready until its cache got sync with kube-apiserver, so the first few readiness probe failed:
```
I0804 12:11:06.052067       1 handler.go:153] antrea-apiserver: GET "/readyz" satisfied by nonGoRestful
I0804 12:11:06.052097       1 pathrecorder.go:240] antrea-apiserver: "/readyz" satisfied by exact match
I0804 12:11:06.052212       1 healthz.go:244] informer-sync check failed: readyz
[-]informer-sync failed: 1 informers not started yet: [*v1.Pod]
...
I0804 12:11:16.043399       1 handler.go:153] antrea-apiserver: GET "/readyz" satisfied by nonGoRestful
I0804 12:11:16.043422       1 pathrecorder.go:240] antrea-apiserver: "/readyz" satisfied by exact match
I0804 12:11:16.044521       1 healthz.go:244] informer-sync check failed: readyz
[-]informer-sync failed: 1 informers not started yet: [*v1.Pod]
```
4. antrea-controller got all Pods from kube-apiserver at 12:11:15 and it took hundreds ms to init the Pod cache:
```
I0804 12:11:15.776949       1 trace.go:205] Trace[60780408]: "Reflector ListAndWatch" name:pkg/mod/k8s.io/client-go@v0.21.0/tools/cache/reflector.go:167 (04-Aug-2021 12:11:02.798) (total time: 12978ms):
I0804 12:11:16.197900       1 shared_informer.go:270] caches populated
I0804 12:11:16.197900       1 controller.go:173] Processing Pod demo-96-mhdphw/scale-ns-mhdphw-rc-1-2jwrj ADD event, labels: map[app-1:scale-1 app-2:scale-2 app-3:scale-3 app-4:scale-4 app-5:scale-5]
I0804 12:11:16.197906       1 shared_informer.go:247] Caches are synced for GroupEntityController
```
5. So antrea-controller was considered Ready after next readiness probe at 12:11:26:
```
I0804 12:11:26.044245       1 handler.go:153] antrea-apiserver: GET "/readyz" satisfied by nonGoRestful
I0804 12:11:26.044282       1 pathrecorder.go:240] antrea-apiserver: "/readyz" satisfied by exact match
I0804 12:11:26.044489       1 httplog.go:89] "HTTP" verb="GET" URI="/readyz" latency="1.47368ms" userAgent="kube-probe/1.20" srcIP="127.0.0.1:57884" resp=200
```
This can also be confirmed from Pod status:
```
    - lastProbeTime: null
      lastTransitionTime: "2021-08-04T12:10:59Z"
      status: "True"
      type: Initialized
    - lastProbeTime: null
      lastTransitionTime: "2021-08-04T12:11:26Z"
      status: "True"
      type: Ready
```
6. kube-proxy took 6s to receive the endpoint update and invoke iptables to redirect the antrea service traffic to antrea-controller, finished at 12:11:33:
```
2021-08-04T12:11:08.106531143Z stderr F I0804 12:11:08.106344       1 trace.go:205] Trace[556545076]: "iptables restore" (04-Aug-2021 12:11:04.688) (total time: 3417ms):
2021-08-04T12:11:08.106571385Z stderr F Trace[556545076]: [3.417356557s] [3.417356557s] END
2021-08-04T12:11:33.924707144Z stderr F I0804 12:11:33.924583       1 trace.go:205] Trace[1355037804]: "iptables restore" (04-Aug-2021 12:11:30.806) (total time: 3117ms):
2021-08-04T12:11:33.924736538Z stderr F Trace[1355037804]: [3.11778385s] [3.11778385s] END
```

So the disconnection last 45s (12:10:48-12:11:33). If the 5 readiness probes happened to be in the time window, the agent Pod would be updated to NotReady and then restarted by Kubernetes.

**To Reproduce**
<!-- Steps to reproduce the behavior. -->
Update antrea-controller Deployment and antrea-agent DaemonSet at same time.
Note It can only reproduce in large scale cluster, for example, a cluster including 50K Pods and 500 Nodes.


**Expected**
<!-- A clear and concise description of what you expected to happen. -->
antrea-agent should update in rolling update fashion to make the container network stable.

**Versions:**
 - Antrea version (Docker image tag). v1.0-v1.2



Pull request description:
antrea-agent readiness probe tolerates longer disconnection
In large-scale clusters, it may take 40~50 seconds for antrea-agent to
reconnect to antrea service after antrea-controller restarts.
antrea-agent shouldn't be reported as NotReady in this scenario,
otherwise DaemonSet controller would restart them at once, as opposed
to rolling update. Set failureThreshold to 8 so it can tolerate 70s of
disconnection.

Signed-off-by: Quan Tian <qtian@vmware.com>

Fixes #2534

The purpose of antrea-agent's readiness probe is to make the status of antrea-agent visible, e.g. the connection with antrea-controller, and it's not for service loadbalancing/failover. So it should be fine to be less aggressive to update its ready status.

Patch:
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index c2f0194a6..5805b70a1 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -4180,7 +4180,7 @@ spec:
           name: api
           protocol: TCP
         readinessProbe:
-          failureThreshold: 5
+          failureThreshold: 8
           httpGet:
             host: localhost
             path: /readyz
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index 53788c248..2ac92a80b 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -4182,7 +4182,7 @@ spec:
           name: api
           protocol: TCP
         readinessProbe:
-          failureThreshold: 5
+          failureThreshold: 8
           httpGet:
             host: localhost
             path: /readyz
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index eccc11946..30ed457d7 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -4180,7 +4180,7 @@ spec:
           name: api
           protocol: TCP
         readinessProbe:
-          failureThreshold: 5
+          failureThreshold: 8
           httpGet:
             host: localhost
             path: /readyz
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index 8677b97d2..8edd9ccda 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -4229,7 +4229,7 @@ spec:
           name: api
           protocol: TCP
         readinessProbe:
-          failureThreshold: 5
+          failureThreshold: 8
           httpGet:
             host: localhost
             path: /readyz
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 64a1b3db7..03d947c71 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -4185,7 +4185,7 @@ spec:
           name: api
           protocol: TCP
         readinessProbe:
-          failureThreshold: 5
+          failureThreshold: 8
           httpGet:
             host: localhost
             path: /readyz
diff --git a/build/yamls/base/agent.yml b/build/yamls/base/agent.yml
index e8a8b32f5..a1a0ec5dc 100644
--- a/build/yamls/base/agent.yml
+++ b/build/yamls/base/agent.yml
@@ -110,7 +110,11 @@ spec:
             initialDelaySeconds: 5
             timeoutSeconds: 5
             periodSeconds: 10
-            failureThreshold: 5
+            # In large-scale clusters, it may take up to 40~50 seconds for the antrea-agent to reconnect to the antrea
+            # Service after the antrea-controller restarts. The antrea-agent shouldn't be reported as NotReady in this
+            # scenario, otherwise the DaemonSet controller would restart all agents at once, as opposed to performing a
+            # rolling update. Set failureThreshold to 8 so it can tolerate 70s of disconnection.
+            failureThreshold: 8
           securityContext:
             # antrea-agent needs to perform sysctl configuration.
             privileged: true
diff --git a/docs/egress.md b/docs/egress.md
index 3ad0ca797..a10d75adc 100644
--- a/docs/egress.md
+++ b/docs/egress.md
@@ -284,7 +284,7 @@ egress-staging       10.10.0.105    1m    node-5
 Now, the packets from the Pods with in the `prod` Namespace to the external
 network will be redirected to the `node-4` Node and SNATed to `10.10.0.104`
 while the packets from the Pods in the `staging` Namespace to the external
-network will be redirected to the `node-5` Node and SNATed to `10.10.0.12`.
+network will be redirected to the `node-5` Node and SNATed to `10.10.0.105`.
 
 In this configuration, if the `node-4` Node powers off, re-configuring
 `10.10.0.104` to another Node or updating the `egressIP` of `egress-prod` to
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index dda8d89c7..3061caeb4 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
@@ -521,15 +522,34 @@ func (c *EgressController) unbindPodEgress(pod, egress string) (string, bool) {
 	return "", false
 }
 
-func (c *EgressController) updateEgressStatus(egress *crdv1a2.Egress, nodeName string) error {
-	if egress.Status.EgressNode == nodeName {
-		return nil
-	}
-	klog.V(2).InfoS("Updating Egress status", "Egress", egress.Name, "oldNode", egress.Status.EgressNode, "newNode", nodeName)
+func (c *EgressController) updateEgressStatus(egress *crdv1a2.Egress, isLocal bool) error {
 	toUpdate := egress.DeepCopy()
-	toUpdate.Status.EgressNode = nodeName
-	if _, err := c.crdClient.CrdV1alpha2().Egresses().UpdateStatus(context.TODO(), toUpdate, metav1.UpdateOptions{}); err != nil {
-		return fmt.Errorf("updating Egress %s status error: %v", egress.Name, err)
+	var updateErr, getErr error
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if isLocal {
+			// Do nothing if the current EgressNode in status is already this Node.
+			if toUpdate.Status.EgressNode == c.nodeName {
+				return nil
+			}
+			toUpdate.Status.EgressNode = c.nodeName
+		} else {
+			// Do nothing if the current EgressNode in status is not this Node.
+			if toUpdate.Status.EgressNode != c.nodeName {
+				return nil
+			}
+			toUpdate.Status.EgressNode = ""
+		}
+		klog.V(2).InfoS("Updating Egress status", "Egress", egress.Name, "oldNode", egress.Status.EgressNode, "newNode", toUpdate.Status.EgressNode)
+		_, updateErr = c.crdClient.CrdV1alpha2().Egresses().UpdateStatus(context.TODO(), toUpdate, metav1.UpdateOptions{})
+		if updateErr != nil && errors.IsConflict(updateErr) {
+			if toUpdate, getErr = c.crdClient.CrdV1alpha2().Egresses().Get(context.TODO(), egress.Name, metav1.GetOptions{}); getErr != nil {
+				return getErr
+			}
+		}
+		// Return the error from UPDATE.
+		return updateErr
+	}); err != nil {
+		return err
 	}
 	klog.V(2).InfoS("Updated Egress status", "Egress", egress.Name)
 	metrics.AntreaEgressStatusUpdates.Inc()
@@ -584,9 +604,6 @@ func (c *EgressController) syncEgress(egressName string) error {
 		if err := c.ipAssigner.AssignIP(egress.Spec.EgressIP); err != nil {
 			return err
 		}
-		if err := c.updateEgressStatus(egress, c.nodeName); err != nil {
-			return err
-		}
 	} else {
 		// Unassign the Egress IP from the local Node if it was assigned by the agent.
 		if err := c.ipAssigner.UnassignIP(egress.Spec.EgressIP); err != nil {
@@ -610,6 +627,10 @@ func (c *EgressController) syncEgress(egressName string) error {
 		eState.mark = mark
 	}
 
+	if err := c.updateEgressStatus(egress, c.localIPDetector.IsLocalIP(egress.Spec.EgressIP)); err != nil {
+		return fmt.Errorf("update Egress %s status error: %v", egressName, err)
+	}
+
 	// Copy the previous ofPorts and Pods. They will be used to identify stale ofPorts and Pods.
 	staleOFPorts := eState.ofPorts.Union(nil)
 	stalePods := eState.pods.Union(nil)
diff --git a/pkg/agent/nodeportlocal/portcache/port_table.go b/pkg/agent/nodeportlocal/portcache/port_table.go
index 644af03ce..19a776913 100644
--- a/pkg/agent/nodeportlocal/portcache/port_table.go
+++ b/pkg/agent/nodeportlocal/portcache/port_table.go
@@ -119,7 +119,7 @@ func (pt *PortTable) AddRule(podIP string, podPort int) (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	if err := pt.PodPortRules.AddRule(nodePort, fmt.Sprintf("%s:%d", podIP, podPort)); err != nil {
+	if err := pt.PodPortRules.AddRule(nodePort, podIP, podPort); err != nil {
 		if err := socket.Close(); err != nil {
 			klog.ErrorS(err, "Unexpected error when closing socket")
 		}
@@ -138,7 +138,7 @@ func (pt *PortTable) DeleteRule(podIP string, podPort int) error {
 	pt.tableLock.Lock()
 	defer pt.tableLock.Unlock()
 	data := pt.getEntryByPodIPPort(podIP, podPort)
-	if err := pt.PodPortRules.DeleteRule(data.NodePort, fmt.Sprintf("%s:%d", podIP, podPort)); err != nil {
+	if err := pt.PodPortRules.DeleteRule(data.NodePort, podIP, podPort); err != nil {
 		return err
 	}
 	if err := data.socket.Close(); err != nil {
diff --git a/pkg/agent/nodeportlocal/rules/iptable_rule.go b/pkg/agent/nodeportlocal/rules/iptable_rule.go
index 7cdb0cf9e..af4bb91e9 100644
--- a/pkg/agent/nodeportlocal/rules/iptable_rule.go
+++ b/pkg/agent/nodeportlocal/rules/iptable_rule.go
@@ -56,16 +56,16 @@ func (ipt *iptablesRules) Init() error {
 // traffic) and OUTPUT chain (for locally-generated traffic). All NPL DNAT rules
 // will be added to this chain.
 func (ipt *iptablesRules) initRules() error {
-	if err := ipt.table.EnsureChain(iptables.NATTable, NodePortLocalChain); err != nil {
+	if err := ipt.table.EnsureChain(iptables.ProtocolIPv4, iptables.NATTable, NodePortLocalChain); err != nil {
 		return err
 	}
 	ruleSpec := []string{
 		"-p", "tcp", "-m", "addrtype", "--dst-type", "LOCAL", "-j", NodePortLocalChain,
 	}
-	if err := ipt.table.EnsureRule(iptables.NATTable, iptables.PreRoutingChain, ruleSpec); err != nil {
+	if err := ipt.table.AppendRule(iptables.ProtocolIPv4, iptables.NATTable, iptables.PreRoutingChain, ruleSpec); err != nil {
 		return err
 	}
-	if err := ipt.table.EnsureRule(iptables.NATTable, iptables.OutputChain, ruleSpec); err != nil {
+	if err := ipt.table.AppendRule(iptables.ProtocolIPv4, iptables.NATTable, iptables.OutputChain, ruleSpec); err != nil {
 		return err
 	}
 	return nil
@@ -78,13 +78,14 @@ func buildRuleForPod(port int, podIP string) []string {
 	}
 }
 
-// AddRule appends a DNAT rule in NodePortLocalChain chain of NAT table
-func (ipt *iptablesRules) AddRule(port int, podIP string) error {
-	rule := buildRuleForPod(port, podIP)
-	if err := ipt.table.EnsureRule(iptables.NATTable, NodePortLocalChain, rule); err != nil {
+// AddRule appends a DNAT rule in NodePortLocalChain chain of NAT table.
+func (ipt *iptablesRules) AddRule(nodePort int, podIP string, podPort int) error {
+	podAddr := fmt.Sprintf("%s:%d", podIP, podPort)
+	rule := buildRuleForPod(nodePort, podAddr)
+	if err := ipt.table.AppendRule(iptables.ProtocolIPv4, iptables.NATTable, NodePortLocalChain, rule); err != nil {
 		return err
 	}
-	klog.Infof("Successfully added DNAT rule for Pod IP %s, port %d", podIP, port)
+	klog.InfoS("Successfully added DNAT rule", "podAddr", podAddr, "nodePort", nodePort)
 	return nil
 }
 
@@ -107,18 +108,19 @@ func (ipt *iptablesRules) AddAllRules(nplList []PodNodePort) error {
 }
 
 // DeleteRule deletes a specific NPL rule from NodePortLocalChain chain
-func (ipt *iptablesRules) DeleteRule(port int, podIP string) error {
-	klog.Infof("Deleting DNAT rule for Pod IP %s, port %d", podIP, port)
-	rule := buildRuleForPod(port, podIP)
-	if err := ipt.table.DeleteRule(iptables.NATTable, NodePortLocalChain, rule); err != nil {
+func (ipt *iptablesRules) DeleteRule(nodePort int, podIP string, podPort int) error {
+	podAddr := fmt.Sprintf("%s:%d", podIP, podPort)
+	rule := buildRuleForPod(nodePort, podAddr)
+	if err := ipt.table.DeleteRule(iptables.ProtocolIPv4, iptables.NATTable, NodePortLocalChain, rule); err != nil {
 		return err
 	}
+	klog.InfoS("Successfully deleted DNAT rule", "podAddr", podAddr, "nodePort", nodePort)
 	return nil
 }
 
 // DeleteAllRules deletes all NPL rules programmed in the node
 func (ipt *iptablesRules) DeleteAllRules() error {
-	exists, err := ipt.table.ChainExists(iptables.NATTable, NodePortLocalChain)
+	exists, err := ipt.table.ChainExists(iptables.ProtocolIPv4, iptables.NATTable, NodePortLocalChain)
 	if err != nil {
 		return fmt.Errorf("failed to check if NodePortLocal chain exists in NAT table: %v", err)
 	}
@@ -128,13 +130,13 @@ func (ipt *iptablesRules) DeleteAllRules() error {
 	ruleSpec := []string{
 		"-p", "tcp", "-m", "addrtype", "--dst-type", "LOCAL", "-j", NodePortLocalChain,
 	}
-	if err := ipt.table.DeleteRule(iptables.NATTable, iptables.PreRoutingChain, ruleSpec); err != nil {
+	if err := ipt.table.DeleteRule(iptables.ProtocolIPv4, iptables.NATTable, iptables.PreRoutingChain, ruleSpec); err != nil {
 		return err
 	}
-	if err := ipt.table.DeleteRule(iptables.NATTable, iptables.OutputChain, ruleSpec); err != nil {
+	if err := ipt.table.DeleteRule(iptables.ProtocolIPv4, iptables.NATTable, iptables.OutputChain, ruleSpec); err != nil {
 		return err
 	}
-	if err := ipt.table.DeleteChain(iptables.NATTable, NodePortLocalChain); err != nil {
+	if err := ipt.table.DeleteChain(iptables.ProtocolIPv4, iptables.NATTable, NodePortLocalChain); err != nil {
 		return err
 	}
 	return nil
diff --git a/pkg/agent/nodeportlocal/rules/rules.go b/pkg/agent/nodeportlocal/rules/rules.go
index a4a70b6d8..7c9bca1ab 100644
--- a/pkg/agent/nodeportlocal/rules/rules.go
+++ b/pkg/agent/nodeportlocal/rules/rules.go
@@ -19,8 +19,8 @@ package rules
 // PodPortRules is an interface to abstract operations on rules for Pods
 type PodPortRules interface {
 	Init() error
-	AddRule(port int, podip string) error
-	DeleteRule(port int, podip string) error
+	AddRule(nodePort int, podIP string, podPort int) error
+	DeleteRule(nodePort int, podIP string, podPort int) error
 	DeleteAllRules() error
 	AddAllRules(nplList []PodNodePort) error
 }
diff --git a/pkg/agent/route/route_linux.go b/pkg/agent/route/route_linux.go
index 06cfca5b3..68aae961d 100644
--- a/pkg/agent/route/route_linux.go
+++ b/pkg/agent/route/route_linux.go
@@ -258,6 +258,7 @@ func (c *Client) syncIPTables() error {
 	var err error
 	v4Enabled := config.IsIPv4Enabled(c.nodeConfig, c.networkConfig.TrafficEncapMode)
 	v6Enabled := config.IsIPv6Enabled(c.nodeConfig, c.networkConfig.TrafficEncapMode)
+
 	c.ipt, err = iptables.New(v4Enabled, v6Enabled)
 	if err != nil {
 		return fmt.Errorf("error creating IPTables instance: %v", err)
@@ -274,11 +275,11 @@ func (c *Client) syncIPTables() error {
 		{iptables.MangleTable, iptables.OutputChain, antreaOutputChain, "Antrea: jump to Antrea output rules"},
 	}
 	for _, rule := range jumpRules {
-		if err := c.ipt.EnsureChain(rule.table, rule.dstChain); err != nil {
+		if err := c.ipt.EnsureChain(iptables.ProtocolDual, rule.table, rule.dstChain); err != nil {
 			return err
 		}
 		ruleSpec := []string{"-j", rule.dstChain, "-m", "comment", "--comment", rule.comment}
-		if err := c.ipt.EnsureRule(rule.table, rule.srcChain, ruleSpec); err != nil {
+		if err := c.ipt.AppendRule(iptables.ProtocolDual, rule.table, rule.srcChain, ruleSpec); err != nil {
 			return err
 		}
 	}
@@ -771,5 +772,5 @@ func (c *Client) DeleteSNATRule(mark uint32) error {
 	}
 	c.markToSNATIP.Delete(mark)
 	snatIP := value.(net.IP)
-	return c.ipt.DeleteRule(iptables.NATTable, antreaPostRoutingChain, c.snatRuleSpec(snatIP, mark))
+	return c.ipt.DeleteRule(iptables.ProtocolDual, iptables.NATTable, antreaPostRoutingChain, c.snatRuleSpec(snatIP, mark))
 }
diff --git a/pkg/agent/util/iptables/iptables.go b/pkg/agent/util/iptables/iptables.go
index 2c6a3fb74..dfc14a090 100644
--- a/pkg/agent/util/iptables/iptables.go
+++ b/pkg/agent/util/iptables/iptables.go
@@ -52,6 +52,15 @@ const (
 
 type Protocol byte
 
+var protocolStrMap = map[Protocol]string{
+	ProtocolIPv4: "IPv4",
+	ProtocolIPv6: "IPv6",
+}
+
+func (p Protocol) String() string {
+	return protocolStrMap[p]
+}
+
 const (
 	ProtocolDual Protocol = iota
 	ProtocolIPv4
@@ -63,20 +72,20 @@ const (
 var restoreWaitSupportedMinVersion = semver.Version{Major: 1, Minor: 6, Patch: 2}
 
 type Client struct {
-	ipts []*iptables.IPTables
+	ipts map[Protocol]*iptables.IPTables
 	// restoreWaitSupported indicates whether iptables-restore (or ip6tables-restore) supports --wait flag.
 	restoreWaitSupported bool
 }
 
 func New(enableIPV4, enableIPV6 bool) (*Client, error) {
-	var ipts []*iptables.IPTables
+	ipts := make(map[Protocol]*iptables.IPTables)
 	var restoreWaitSupported bool
 	if enableIPV4 {
 		ipt, err := iptables.New()
 		if err != nil {
 			return nil, fmt.Errorf("error creating IPTables instance: %v", err)
 		}
-		ipts = append(ipts, ipt)
+		ipts[ProtocolIPv4] = ipt
 		restoreWaitSupported = isRestoreWaitSupported(ipt)
 	}
 	if enableIPV6 {
@@ -84,7 +93,7 @@ func New(enableIPV4, enableIPV6 bool) (*Client, error) {
 		if err != nil {
 			return nil, fmt.Errorf("error creating IPTables instance for IPv6: %v", err)
 		}
-		ipts = append(ipts, ip6t)
+		ipts[ProtocolIPv6] = ip6t
 		if !restoreWaitSupported {
 			restoreWaitSupported = isRestoreWaitSupported(ip6t)
 		}
@@ -99,9 +108,12 @@ func isRestoreWaitSupported(ipt *iptables.IPTables) bool {
 }
 
 // EnsureChain checks if target chain already exists, creates it if not.
-func (c *Client) EnsureChain(table string, chain string) error {
-	for idx := range c.ipts {
-		ipt := c.ipts[idx]
+func (c *Client) EnsureChain(protocol Protocol, table string, chain string) error {
+	for p := range c.ipts {
+		ipt := c.ipts[p]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
 		exists, err := ipt.ChainExists(table, chain)
 		if err != nil {
 			return fmt.Errorf("error checking if chain %s exists in table %s: %v", chain, table, err)
@@ -112,48 +124,56 @@ func (c *Client) EnsureChain(table string, chain string) error {
 		if err := ipt.NewChain(table, chain); err != nil {
 			return fmt.Errorf("error creating chain %s in table %s: %v", chain, table, err)
 		}
-		klog.V(2).Infof("Created chain %s in table %s", chain, table)
+		klog.V(2).InfoS("Created a chain", "chain", chain, "table", table, "protocol", p)
 	}
 	return nil
 }
 
-// ChainExists checks if a chain already exists in a table
-func (c *Client) ChainExists(table string, chain string) (bool, error) {
-	for idx := range c.ipts {
-		exists, err := c.ipts[idx].ChainExists(table, chain)
+// ChainExists checks if target chain already exists in a table
+func (c *Client) ChainExists(protocol Protocol, table string, chain string) (bool, error) {
+	for p := range c.ipts {
+		ipt := c.ipts[p]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
+		exists, err := ipt.ChainExists(table, chain)
 		if err != nil {
 			return false, fmt.Errorf("error checking if chain %s exists in table %s: %v", chain, table, err)
 		}
-		if exists {
-			return true, nil
+		if !exists {
+			return false, nil
 		}
+		klog.V(2).InfoS("A chain exists", "chain", chain, "table", table, "protocol", p)
 	}
-	return false, nil
+	return true, nil
 }
 
-// EnsureRule checks if target rule already exists, appends it if not.
-func (c *Client) EnsureRule(table string, chain string, ruleSpec []string) error {
-	for idx := range c.ipts {
-		ipt := c.ipts[idx]
+// AppendRule checks if target rule already exists with the protocol, appends it if not.
+func (c *Client) AppendRule(protocol Protocol, table string, chain string, ruleSpec []string) error {
+	for p := range c.ipts {
+		ipt := c.ipts[p]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
 		exist, err := ipt.Exists(table, chain, ruleSpec...)
 		if err != nil {
 			return fmt.Errorf("error checking if rule %v exists in table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
 		if exist {
-			return nil
+			continue
 		}
 		if err := ipt.Append(table, chain, ruleSpec...); err != nil {
 			return fmt.Errorf("error appending rule %v to table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
+		klog.V(2).InfoS("Appended a rule", "rule", ruleSpec, "table", table, "chain", chain, "protocol", p)
 	}
-	klog.V(2).Infof("Appended rule %v to table %s chain %s", ruleSpec, table, chain)
 	return nil
 }
 
 // InsertRule checks if target rule already exists, inserts it if not.
 func (c *Client) InsertRule(protocol Protocol, table string, chain string, ruleSpec []string) error {
-	for idx := range c.ipts {
-		ipt := c.ipts[idx]
+	for p := range c.ipts {
+		ipt := c.ipts[p]
 		if !matchProtocol(ipt, protocol) {
 			continue
 		}
@@ -162,13 +182,13 @@ func (c *Client) InsertRule(protocol Protocol, table string, chain string, ruleS
 			return fmt.Errorf("error checking if rule %v exists in table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
 		if exist {
-			return nil
+			continue
 		}
 		if err := ipt.Insert(table, chain, 1, ruleSpec...); err != nil {
 			return fmt.Errorf("error inserting rule %v to table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
+		klog.V(2).InfoS("Inserted a rule", "rule", ruleSpec, "table", table, "chain", chain)
 	}
-	klog.V(2).Infof("Inserted rule %v to table %s chain %s", ruleSpec, table, chain)
 	return nil
 }
 
@@ -185,46 +205,59 @@ func matchProtocol(ipt *iptables.IPTables, protocol Protocol) bool {
 }
 
 // DeleteRule checks if target rule already exists, deletes the rule if found.
-func (c *Client) DeleteRule(table string, chain string, ruleSpec []string) error {
-	for idx := range c.ipts {
-		exist, err := c.ipts[idx].Exists(table, chain, ruleSpec...)
+func (c *Client) DeleteRule(protocol Protocol, table string, chain string, ruleSpec []string) error {
+	for p := range c.ipts {
+		ipt := c.ipts[p]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
+		exist, err := ipt.Exists(table, chain, ruleSpec...)
 		if err != nil {
 			return fmt.Errorf("error checking if rule %v exists in table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
 		if !exist {
-			return nil
+			continue
 		}
-		if err := c.ipts[idx].Delete(table, chain, ruleSpec...); err != nil {
+		if err := ipt.Delete(table, chain, ruleSpec...); err != nil {
 			return fmt.Errorf("error deleting rule %v from table %s chain %s: %v", ruleSpec, table, chain, err)
 		}
-		klog.V(2).Infof("Deleted rule %v from table %s chain %s", ruleSpec, table, chain)
+		klog.V(2).InfoS("Deleted a rule", "rule", ruleSpec, "table", table, "chain", chain, "protocol", p)
 	}
 	return nil
 }
 
-// DeleteChain deletes all rules from a chain in a table and then delete the chain
-func (c *Client) DeleteChain(table string, chain string) error {
-	for idx := range c.ipts {
-		err := c.ipts[idx].ClearChain(table, chain)
+// DeleteChain deletes all rules from a chain in a table and then delete the chain.
+func (c *Client) DeleteChain(protocol Protocol, table string, chain string) error {
+	for p := range c.ipts {
+		ipt := c.ipts[p]
+		if !matchProtocol(ipt, protocol) {
+			continue
+		}
+		exists, err := ipt.ChainExists(table, chain)
 		if err != nil {
+			return fmt.Errorf("error checking if chain %s exists in table %s: %v", chain, table, err)
+		}
+		if !exists {
+			continue
+		}
+		if err = ipt.ClearChain(table, chain); err != nil {
 			return fmt.Errorf("error clearing rules from table %s chain %s: %v", table, chain, err)
 		}
-		err = c.ipts[idx].DeleteChain(table, chain)
-		if err != nil {
+		if err = ipt.DeleteChain(table, chain); err != nil {
 			return fmt.Errorf("error deleting chain %s from table %s: %v", chain, table, err)
 		}
-		klog.V(2).Infof("Deleted chain %s from table %s", chain, table)
+		klog.V(2).InfoS("Deleted a chain", "chain", chain, "table", table, "protocol", p)
 	}
 	return nil
 }
 
-// ListRules lists all rules from a chain in a table
+// ListRules lists all rules from a chain in a table.
 func (c *Client) ListRules(table string, chain string) ([]string, error) {
 	var allRules []string
-	for idx := range c.ipts {
-		rules, err := c.ipts[idx].List(table, chain)
+	for p := range c.ipts {
+		rules, err := c.ipts[p].List(table, chain)
 		if err != nil {
-			return rules, fmt.Errorf("error getting rules from table %s chain %s: %v", table, chain, err)
+			return rules, fmt.Errorf("error getting rules from table %s chain %s protocol %s: %v", table, chain, p, err)
 		}
 		allRules = append(allRules, rules...)
 	}
@@ -265,7 +298,7 @@ func (c *Client) Restore(data []byte, flush bool, useIPv6 bool) error {
 		defer unlockFunc()
 	}
 	if err := cmd.Run(); err != nil {
-		klog.Errorf("Failed to execute %s: %v\nstdin:\n%s\nstderr:\n%s", iptablesCmd, err, data, stderr)
+		klog.ErrorS(err, "Failed to execute iptables command", "iptablesCmd", iptablesCmd, "stdin", data, "stderr", stderr)
 		return fmt.Errorf("error executing %s: %v", iptablesCmd, err)
 	}
 	return nil
@@ -274,9 +307,9 @@ func (c *Client) Restore(data []byte, flush bool, useIPv6 bool) error {
 // Save calls iptables-saves to dump chains and tables in iptables.
 func (c *Client) Save() ([]byte, error) {
 	var output []byte
-	for idx := range c.ipts {
+	for p := range c.ipts {
 		var cmd string
-		ipt := c.ipts[idx]
+		ipt := c.ipts[p]
 		switch ipt.Proto() {
 		case iptables.ProtocolIPv6:
 			cmd = "ip6tables-save"


Test patch:
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 95502c359..31e54b7cf 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -16,6 +16,7 @@ package egress
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"reflect"
 	"testing"
@@ -24,10 +25,12 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	k8stesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/util/workqueue"
 
 	ipassignertest "antrea.io/antrea/pkg/agent/controller/egress/ipassigner/testing"
@@ -38,7 +41,6 @@ import (
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	"antrea.io/antrea/pkg/client/clientset/versioned"
-	"antrea.io/antrea/pkg/client/clientset/versioned/fake"
 	fakeversioned "antrea.io/antrea/pkg/client/clientset/versioned/fake"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions"
 	"antrea.io/antrea/pkg/util/k8s"
@@ -98,7 +100,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 	mockRouteClient := routetest.NewMockInterface(controller)
 	mockIPAssigner := ipassignertest.NewMockIPAssigner(controller)
 
-	clientset := &fake.Clientset{}
+	clientset := &fakeversioned.Clientset{}
 	crdClient := fakeversioned.NewSimpleClientset(initObjects...)
 	crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, 0)
 	egressInformer := crdInformerFactory.Crd().V1alpha2().Egresses()
@@ -114,6 +116,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 	egressController := &EgressController{
 		ofClient:             mockOFClient,
 		routeClient:          mockRouteClient,
+		crdClient:            crdClient,
 		antreaClientProvider: &antreaClientGetter{clientset},
 		egressInformer:       egressInformer.Informer(),
 		egressLister:         egressInformer.Lister(),
@@ -148,6 +151,7 @@ func TestSyncEgress(t *testing.T) {
 		existingEgressGroup *cpv1b2.EgressGroup
 		newEgressGroup      *cpv1b2.EgressGroup
 		newLocalIPs         sets.String
+		expectedEgresses    []*crdv1a2.Egress
 		expectedCalls       func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner)
 	}{
 		{
@@ -175,6 +179,12 @@ func TestSyncEgress(t *testing.T) {
 				},
 			},
 			newLocalIPs: sets.NewString(),
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -218,6 +228,13 @@ func TestSyncEgress(t *testing.T) {
 				},
 			},
 			newLocalIPs: sets.NewString(fakeRemoteEgressIP1),
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeRemoteEgressIP1},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
@@ -258,6 +275,13 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP2},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -303,6 +327,12 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeRemoteEgressIP1},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -346,6 +376,13 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
@@ -387,6 +424,18 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressB", UID: "uidB"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP2},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -426,6 +475,18 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
+			expectedEgresses: []*crdv1a2.Egress{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "egressB", UID: "uidB"},
+					Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+					Status:     crdv1a2.EgressStatus{EgressNode: fakeNode},
+				},
+			},
 			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -470,6 +531,11 @@ func TestSyncEgress(t *testing.T) {
 			// Call it one more time to ensure it's idempotent, no extra datapath calls are supposed to be made.
 			err = c.syncEgress(tt.newEgress.Name)
 			assert.NoError(t, err)
+			for _, expectedEgress := range tt.expectedEgresses {
+				gotEgress, err := c.crdClient.CrdV1alpha2().Egresses().Get(context.TODO(), expectedEgress.Name, metav1.GetOptions{})
+				require.NoError(t, err)
+				assert.Equal(t, expectedEgress, gotEgress)
+			}
 		})
 	}
 }
@@ -624,3 +690,90 @@ func addPodInterface(ifaceStore interfacestore.InterfaceStore, podNamespace, pod
 		OVSPortConfig:            &interfacestore.OVSPortConfig{OFPort: ofPort},
 	})
 }
+
+func TestUpdateEgressStatus(t *testing.T) {
+	getError := fmt.Errorf("fake get Egress error")
+	updateConflictError := &errors.StatusError{ErrStatus: metav1.Status{Reason: metav1.StatusReasonConflict, Message: "update Egress conflict"}}
+	updateError := fmt.Errorf("update Egress error")
+	tests := []struct {
+		name                 string
+		updateErrorNum       int
+		updateError          error
+		getErrorNum          int
+		expectedUpdateCalled int
+		expectedGetCalled    int
+		expectedError        error
+	}{
+		{
+			name:                 "succeed immediately",
+			updateErrorNum:       0,
+			updateError:          nil,
+			getErrorNum:          0,
+			expectedUpdateCalled: 1,
+			expectedGetCalled:    0,
+			expectedError:        nil,
+		},
+		{
+			name:                 "succeed after one update conflict failure",
+			updateErrorNum:       1,
+			updateError:          updateConflictError,
+			getErrorNum:          0,
+			expectedUpdateCalled: 2,
+			expectedGetCalled:    1,
+			expectedError:        nil,
+		},
+		{
+			name:                 "fail after one update failure",
+			updateErrorNum:       1,
+			updateError:          updateError,
+			getErrorNum:          0,
+			expectedUpdateCalled: 1,
+			expectedGetCalled:    0,
+			expectedError:        updateError,
+		},
+		{
+			name:                 "fail after one update conflict failure and one get failure",
+			updateErrorNum:       1,
+			updateError:          updateConflictError,
+			getErrorNum:          1,
+			expectedUpdateCalled: 1,
+			expectedGetCalled:    1,
+			expectedError:        getError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			egress := crdv1a2.Egress{
+				ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA", ResourceVersion: "fake-ResourceVersion"},
+				Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+			}
+			fakeClient := &fakeversioned.Clientset{}
+			getCalled := 0
+			fakeClient.AddReactor("get", "egresses", func(action k8stesting.Action) (bool, runtime.Object, error) {
+				getCalled += 1
+				if getCalled <= tt.getErrorNum {
+					return true, nil, getError
+				}
+				return true, &egress, nil
+			})
+			updateCalled := 0
+			fakeClient.AddReactor("update", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+				updateCalled += 1
+				if updateCalled <= tt.updateErrorNum {
+					return true, nil, tt.updateError
+				}
+				return true, &egress, nil
+			})
+
+			c := &EgressController{crdClient: fakeClient, nodeName: fakeNode}
+			_, err := c.crdClient.CrdV1alpha2().Egresses().Create(context.TODO(), &egress, metav1.CreateOptions{})
+			assert.NoError(t, err)
+			err = c.updateEgressStatus(&egress, true)
+			if err != tt.expectedError {
+				t.Errorf("Update Egress error not match, got: %v, expected: %v", err, tt.expectedError)
+			}
+			assert.Equal(t, tt.expectedGetCalled, getCalled, "Get called num not match")
+			assert.Equal(t, tt.expectedUpdateCalled, updateCalled, "Update called num not match")
+		})
+	}
+}
diff --git a/pkg/agent/nodeportlocal/npl_agent_test.go b/pkg/agent/nodeportlocal/npl_agent_test.go
index ea9aae035..40c8c7abf 100644
--- a/pkg/agent/nodeportlocal/npl_agent_test.go
+++ b/pkg/agent/nodeportlocal/npl_agent_test.go
@@ -191,8 +191,8 @@ 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()).AnyTimes()
-	mockIPTables.EXPECT().DeleteRule(gomock.Any(), gomock.Any()).AnyTimes()
+	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)
diff --git a/pkg/agent/nodeportlocal/rules/testing/mock_rules.go b/pkg/agent/nodeportlocal/rules/testing/mock_rules.go
index b0ce5f010..da1d7a03e 100644
--- a/pkg/agent/nodeportlocal/rules/testing/mock_rules.go
+++ b/pkg/agent/nodeportlocal/rules/testing/mock_rules.go
@@ -63,17 +63,17 @@ func (mr *MockPodPortRulesMockRecorder) AddAllRules(arg0 interface{}) *gomock.Ca
 }
 
 // AddRule mocks base method
-func (m *MockPodPortRules) AddRule(arg0 int, arg1 string) error {
+func (m *MockPodPortRules) AddRule(arg0 int, arg1 string, arg2 int) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "AddRule", arg0, arg1)
+	ret := m.ctrl.Call(m, "AddRule", arg0, arg1, arg2)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // AddRule indicates an expected call of AddRule
-func (mr *MockPodPortRulesMockRecorder) AddRule(arg0, arg1 interface{}) *gomock.Call {
+func (mr *MockPodPortRulesMockRecorder) AddRule(arg0, arg1, arg2 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddRule", reflect.TypeOf((*MockPodPortRules)(nil).AddRule), arg0, arg1)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddRule", reflect.TypeOf((*MockPodPortRules)(nil).AddRule), arg0, arg1, arg2)
 }
 
 // DeleteAllRules mocks base method
@@ -91,17 +91,17 @@ func (mr *MockPodPortRulesMockRecorder) DeleteAllRules() *gomock.Call {
 }
 
 // DeleteRule mocks base method
-func (m *MockPodPortRules) DeleteRule(arg0 int, arg1 string) error {
+func (m *MockPodPortRules) DeleteRule(arg0 int, arg1 string, arg2 int) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DeleteRule", arg0, arg1)
+	ret := m.ctrl.Call(m, "DeleteRule", arg0, arg1, arg2)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // DeleteRule indicates an expected call of DeleteRule
-func (mr *MockPodPortRulesMockRecorder) DeleteRule(arg0, arg1 interface{}) *gomock.Call {
+func (mr *MockPodPortRulesMockRecorder) DeleteRule(arg0, arg1, arg2 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteRule", reflect.TypeOf((*MockPodPortRules)(nil).DeleteRule), arg0, arg1)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteRule", reflect.TypeOf((*MockPodPortRules)(nil).DeleteRule), arg0, arg1, arg2)
 }
 
 // Init mocks base method
diff --git a/test/e2e/egress_test.go b/test/e2e/egress_test.go
index 29c2928fc..5394beb8f 100644
--- a/test/e2e/egress_test.go
+++ b/test/e2e/egress_test.go
@@ -169,13 +169,23 @@ ip netns exec %[1]s /agnhost netexec
 	assertClientIP(localPod, egressNodeIP)
 	assertClientIP(remotePod, egressNodeIP)
 
+	var err error
+	err = wait.Poll(time.Millisecond*100, time.Second, func() (bool, error) {
+		egress, err = data.crdClient.CrdV1alpha2().Egresses().Get(context.TODO(), egress.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		return egress.Status.EgressNode == egressNode, nil
+	})
+	assert.NoError(t, err, "Egress failed to reach expected status")
+
 	t.Log("Updating the Egress's AppliedTo to remotePod only")
 	egress.Spec.AppliedTo = v1alpha2.AppliedTo{
 		PodSelector: &metav1.LabelSelector{
 			MatchLabels: map[string]string{"antrea-e2e": remotePod},
 		},
 	}
-	egress, err := data.crdClient.CrdV1alpha2().Egresses().Update(context.TODO(), egress, metav1.UpdateOptions{})
+	egress, err = data.crdClient.CrdV1alpha2().Egresses().Update(context.TODO(), egress, metav1.UpdateOptions{})
 	if err != nil {
 		t.Fatalf("Failed to update Egress %v: %v", egress, err)
 	}
@@ -342,7 +352,7 @@ func testEgressUpdateEgressIP(t *testing.T, data *TestData) {
 			toUpdate.Spec.ExternalIPPool = newPool.Name
 			toUpdate.Spec.EgressIP = newIP
 			egress, err = data.crdClient.CrdV1alpha2().Egresses().Update(context.TODO(), toUpdate, metav1.UpdateOptions{})
-			require.NoError(t, err, "Failed to delete Egress")
+			require.NoError(t, err, "Failed to update Egress")
 
 			_, err = data.checkEgressState(egress.Name, newIP, tt.newNode, "", time.Second)
 			require.NoError(t, err)