Redundant Openflow messages are sent when syncing an updated group to OVS

[hongliangl]

Describe the bug

For example, when syncing an ClusterIP Service whose Endpoints has been changed,

Function in the following is called in pkg/agent/openflow/client.go, we can see that variable group is returned by serviceEndpointGroup , then call the method Add() to sync the group to OVS.

func (c *client) InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error {
	c.replayMutex.RLock()
	defer c.replayMutex.RUnlock()

	group := c.featureService.serviceEndpointGroup(groupID, withSessionAffinity, endpoints...)
	if err := group.Add(); err != nil {
		return fmt.Errorf("error when installing Service Endpoints Group: %w", err)
	}
	c.featureService.groupCache.Store(groupID, group)
	return nil
}

We can see the code of method Add() in pkg/ovs/openflow/ofctrl_group.go:

func (g *ofGroup) Add() error {
	return g.ofctrl.Install()
}

We can see the the method Install() of g.ofctrl in lib ofnet https://github.com/antrea-io/ofnet/blob/716e8ccd4ba3713d3b38ff2f808e698749d9de79/ofctrl/fgraphGroup.go#L76

func (self *Group) Install() error {
	command := openflow15.OFPGC_ADD
	if self.isInstalled {
		command = openflow15.OFPGC_MODIFY
	}
	groupMod := self.getGroupModMessage(command)

	if err := self.Switch.Send(groupMod); err != nil {
		return err
	}

	// Mark it as installed
	self.isInstalled = true

	return nil
}

However, in serviceEndpointGroup which is used to generate group in /pkg/agent/openflow/pipeline.go, we can see that when adding a new bucket, then Done() is called.

func (f *featureService) serviceEndpointGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints ...proxy.Endpoint) binding.Group {
	group := f.bridge.CreateGroup(groupID).ResetBuckets()
	var resubmitTableID uint8
	if withSessionAffinity {
		resubmitTableID = ServiceLBTable.GetID()
	} else {
		resubmitTableID = EndpointDNATTable.GetID()
	}
	for _, endpoint := range endpoints {
		endpointPort, _ := endpoint.Port()
		endpointIP := net.ParseIP(endpoint.IP())
		portVal := util.PortToUint16(endpointPort)
		ipProtocol := getIPProtocol(endpointIP)

		if ipProtocol == binding.ProtocolIP {
			ipVal := binary.BigEndian.Uint32(endpointIP.To4())
			group = group.Bucket().Weight(100).
				LoadToRegField(EndpointIPField, ipVal).
				LoadToRegField(EndpointPortField, uint32(portVal)).
				ResubmitToTable(resubmitTableID).
				Done()
		} else if ipProtocol == binding.ProtocolIPv6 {
			ipVal := []byte(endpointIP)
			group = group.Bucket().Weight(100).
				LoadXXReg(EndpointIP6Field.GetRegID(), ipVal).
				LoadToRegField(EndpointPortField, uint32(portVal)).
				ResubmitToTable(resubmitTableID).
				Done()
		}
	}
	return group
}

We can see the code in Done() in /pkg/ovs/openflow/ofctrl_group.go, AddBuckets is called.

func (b *bucketBuilder) Done() Group {
	b.group.ofctrl.AddBuckets(b.bucket)
	return b.group
}

AddBuckets is defined in lib ofnet https://github.com/antrea-io/ofnet/blob/716e8ccd4ba3713d3b38ff2f808e698749d9de79/ofctrl/fgraphGroup.go#L58:

func (self *Group) AddBuckets(buckets ...*openflow15.Bucket) {
	if self.Buckets == nil {
		self.Buckets = make([]*openflow15.Bucket, 0)
	}
	self.Buckets = append(self.Buckets, buckets...)
	if self.isInstalled {
		self.Install()
	}
}

We can see that if self.isInstalled is true (this means that when updating the group), Install() is called.

Take a simple example, when syncing a updated Service which has 3 Endpoints, there are

• Message to install the corresponding group which has bucket 0 (triggered by Done() to generate bucket 0)
• Message to install the corresponding group which has bucket 0, 1 (triggered by Done() to generate bucket 1)
• Message to install the corresponding group which has bucket 0, 1, 2 (triggered by Done() to generate bucket 2)
• Message to install the corresponding group which has bucket 0, 1, 2 (triggered by Add() to install the group)

Only the last Openflow is needed when syncing an updated group to OVS. To fix the issue, the code in Done() could be:

func (b *bucketBuilder) Done() Group {
	b.group.ofctrl.Buckets = append(b.group.ofctrl.Buckets, b.bucket)
	return b.group
}

[tnqn]

@hongliangl nice catch!