Fix OVS "flow" replay for groups

pkg/agent/agent.go

@@ -281,6 +281,12 @@ func persistRoundNum(num uint64, bridgeClient ovsconfig.OVSBridgeClient, interva
 // agent restarts (with the agent crashing before step 4 can be completed). With the sequence
 // described above, We guarantee that at most two rounds of flows exist in the switch at any given
 // time.
+// Note that at the moment we assume that all OpenFlow groups are deleted every time there is an
+// Antrea Agent restart. This allows us to add the necessary groups without having to worry about
+// the operation failing because a (stale) group with the same ID already exists in OVS. This
+// assumption is currently guaranteed by the ofnet implementation:
+// https://github.com/wenyingd/ofnet/blob/14a78b27ef8762e45a0cfc858c4d07a4572a99d5/ofctrl/fgraphSwitch.go#L57-L62
+// All previous groups have been deleted by the time the call to i.ofClient.Initialize returns.
 func (i *Initializer) initOpenFlowPipeline() error {
 	roundInfo := getRoundInfo(i.ovsBridgeClient)
 
@@ -427,7 +433,7 @@ func (i *Initializer) FlowRestoreComplete() error {
 	if err != nil {
 		if err == wait.ErrWaitTimeout {
 			// This could happen if the method is triggered by OVS disconnection event, in which OVS doesn't restart.
-			klog.Info("flow-restore-wait was not true, skip cleaning up it")
+			klog.Info("flow-restore-wait was not true, skip cleaning it up")
 			return nil
 		}
 		return err

pkg/agent/openflow/client.go

@@ -753,8 +753,10 @@ func (c *client) ReplayFlows() {
 		return true
 	}
 
-	c.groupCache.Range(func(id, gEntry interface{}) bool {
-		if err := gEntry.(binding.Group).Add(); err != nil {
+	c.groupCache.Range(func(id, value interface{}) bool {
+		group := value.(binding.Group)
+		group.Reset()
+		if err := group.Add(); err != nil {
 			klog.Errorf("Error when replaying cached group %d: %v", id, err)
 		}
 		return true

pkg/ovs/openflow/ofctrl_bridge.go

@@ -150,7 +150,7 @@ type OFBridge struct {
 
 func (b *OFBridge) CreateGroup(id GroupIDType) Group {
 	ofctrlGroup, err := b.ofSwitch.NewGroup(uint32(id), ofctrl.GroupSelect)
-	if err != nil {
+	if err != nil { // group already exists
 		ofctrlGroup = b.ofSwitch.GetGroup(uint32(id))
 	}
 	g := &ofGroup{bridge: b, ofctrl: ofctrlGroup}

pkg/ovs/openflow/ofctrl_group.go

@@ -26,8 +26,18 @@ type ofGroup struct {
 	bridge *OFBridge
 }
 
+// Reset creates a new ofctrl.Group object for the updated ofSwitch. The
+// ofSwitch keeps a list of all Group objects, so this operation is
+// needed. Reset() should be called before replaying the Group to OVS.
 func (g *ofGroup) Reset() {
-	g.ofctrl.Switch = g.bridge.ofSwitch
+	// An error ("group already exists") is not possible here since we are
+	// using a new instance of ofSwitch and re-creating a group which was
+	// created successfully before. There will be no duplicate group IDs. If
+	// something is wrong and there is an error, g.ofctrl will be set to nil
+	// and the Agent will crash later.
+	newGroup, _ := g.bridge.ofSwitch.NewGroup(g.ofctrl.ID, g.ofctrl.GroupType)
+	newGroup.Buckets = g.ofctrl.Buckets
+	g.ofctrl = newGroup
 }
 
 func (g *ofGroup) Add() error {

test/e2e/connectivity_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 
 	"github.com/vmware-tanzu/antrea/pkg/agent/config"
@@ -327,8 +328,9 @@ func TestOVSRestartSameNode(t *testing.T) {
 }
 
 // TestOVSFlowReplay checks that when OVS restarts unexpectedly the Antrea agent takes care of
-// replaying flows. More precisely this test checks that Pod connectivity still works after deleting
-// the flows and force-restarting the OVS dameons.
+// replaying flows. More precisely this test checks that we have the same number of flows and groups
+// after deleting them and force-restarting the OVS daemons. We also make sure that Pod connectivity
+// still works.
 func TestOVSFlowReplay(t *testing.T) {
 	skipIfProviderIs(t, "kind", "stopping OVS daemons create connectivity issues")
 	data, err := setupTest(t)
@@ -360,15 +362,45 @@ func TestOVSFlowReplay(t *testing.T) {
 	}
 	t.Logf("The Antrea Pod for Node '%s' is '%s'", workerNode, antreaPodName)
 
-	t.Logf("Deleting flows and restarting OVS daemons on Node '%s'", workerNode)
-	delFlows := func() {
+	countFlows := func() int {
+		cmd := []string{"ovs-ofctl", "dump-flows", defaultBridgeName}
+		stdout, stderr, err := data.runCommandFromPod(antreaNamespace, antreaPodName, ovsContainerName, cmd)
+		if err != nil {
+			t.Fatalf("error when dumping flows: <%v>, err: <%v>", stderr, err)
+		}
+		count := strings.Count(stdout, "\n")
+		t.Logf("Counted %d flow in OVS bridge '%s' for Node '%s'", count, defaultBridgeName, workerNode)
+		return count
+	}
+	countGroups := func() int {
+		cmd := []string{"ovs-ofctl", "dump-groups", defaultBridgeName}
+		stdout, stderr, err := data.runCommandFromPod(antreaNamespace, antreaPodName, ovsContainerName, cmd)
+		if err != nil {
+			t.Fatalf("error when dumping groups: <%v>, err: <%v>", stderr, err)
+		}
+		count := strings.Count(stdout, "\n")
+		t.Logf("Counted %d group in OVS bridge '%s' for Node '%s'", count, defaultBridgeName, workerNode)
+		return count
+	}
+
+	numFlows1, numGroups1 := countFlows(), countGroups()
+
+	// This is necessary because "ovs-ctl restart" saves and restores OpenFlow flows for the
+	// bridge. An alternative may be to kill the antrea-ovs container running on that Node.
+	t.Logf("Deleting flows / groups and restarting OVS daemons on Node '%s'", workerNode)
+	delFlowsAndGroups := func() {
 		cmd := []string{"ovs-ofctl", "del-flows", defaultBridgeName}
 		_, stderr, err := data.runCommandFromPod(antreaNamespace, antreaPodName, ovsContainerName, cmd)
 		if err != nil {
 			t.Fatalf("error when deleting flows: <%v>, err: <%v>", stderr, err)
 		}
+		cmd = []string{"ovs-ofctl", "del-groups", defaultBridgeName}
+		_, stderr, err = data.runCommandFromPod(antreaNamespace, antreaPodName, ovsContainerName, cmd)
+		if err != nil {
+			t.Fatalf("error when deleting groups: <%v>, err: <%v>", stderr, err)
+		}
 	}
-	delFlows()
+	delFlowsAndGroups()
 	restartCmd := []string{"/usr/share/openvswitch/scripts/ovs-ctl", "--system-id=random", "restart", "--db-file=/var/run/openvswitch/conf.db"}
 	if stdout, stderr, err := data.runCommandFromPod(antreaNamespace, antreaPodName, ovsContainerName, restartCmd); err != nil {
 		t.Fatalf("Error when restarting OVS with ovs-ctl: %v - stdout: %s - stderr: %s", err, stdout, stderr)
@@ -378,6 +410,10 @@ func TestOVSFlowReplay(t *testing.T) {
 	// interval.
 	t.Logf("Running second ping mesh to check that flows have been restored")
 	data.runPingMesh(t, podNames)
+
+	numFlows2, numGroups2 := countFlows(), countGroups()
+	assert.Equal(t, numFlows1, numFlows2, "Mismatch in OVS flow count after flow replay")
+	assert.Equal(t, numGroups1, numGroups2, "Mismatch in OVS group count after flow replay")
 }
 
 // TestPingLargeMTU verifies that fragmented ICMP packets are handled correctly. Until OVS 2.12.0,