underlay: sync NetworkManager IP config to OVS bridge

pkg/daemon/controller.go

@@ -277,7 +277,7 @@ func (c *Controller) initProviderNetwork(pn *kubeovnv1.ProviderNetwork, node *v1
 	var mtu int
 	var err error
 	klog.V(3).Infof("ovs init provider network %s", pn.Name)
-	if mtu, err = ovsInitProviderNetwork(pn.Name, nic, pn.Spec.ExchangeLinkName, c.config.MacLearningFallback); err != nil {
+	if mtu, err = c.ovsInitProviderNetwork(pn.Name, nic, pn.Spec.ExchangeLinkName, c.config.MacLearningFallback); err != nil {
 		if oldLen := len(node.Labels); oldLen != 0 {
 			delete(node.Labels, fmt.Sprintf(util.ProviderNetworkReadyTemplate, pn.Name))
 			delete(node.Labels, fmt.Sprintf(util.ProviderNetworkInterfaceTemplate, pn.Name))
@@ -391,15 +391,15 @@ func (c *Controller) cleanProviderNetwork(pn *kubeovnv1.ProviderNetwork, node *v
 		return err
 	}
 
-	if err = ovsCleanProviderNetwork(pn.Name); err != nil {
+	if err = c.ovsCleanProviderNetwork(pn.Name); err != nil {
 		return err
 	}
 
 	return nil
 }
 
 func (c *Controller) handleDeleteProviderNetwork(pn *kubeovnv1.ProviderNetwork) error {
-	if err := ovsCleanProviderNetwork(pn.Name); err != nil {
+	if err := c.ovsCleanProviderNetwork(pn.Name); err != nil {
 		return err
 	}
 

pkg/daemon/controller_linux.go

@@ -38,6 +38,8 @@ type ControllerRuntime struct {
 	k8sipsets        k8sipset.Interface
 	ipsets           map[string]*ipsets.IPSets
 	gwCounters       map[string]*util.GwIPtableCounters
+
+	nmSyncer *networkManagerSyncer
 }
 
 func evalCommandSymlinks(cmd string) (string, error) {
@@ -113,6 +115,9 @@ func (c *Controller) initRuntime() error {
 		c.k8siptables[kubeovnv1.ProtocolIPv6] = k8siptables.New(c.k8sExec, k8siptables.ProtocolIPv6)
 	}
 
+	c.nmSyncer = newNetworkManagerSyncer()
+	c.nmSyncer.Run(c.transferAddrsAndRoutes)
+
 	return nil
 }
 

pkg/daemon/init.go

@@ -40,10 +40,6 @@ func InitOVSBridges() (map[string]string, error) {
 					return nil, fmt.Errorf("failed to check vendor of port %s: %v", port, err)
 				}
 				if ok {
-					klog.Infof("config provider nic %s on bridge %s", port, brName)
-					if _, err = configProviderNic(port, brName); err != nil {
-						return nil, err
-					}
 					mappings[port] = brName
 				}
 			}
@@ -97,11 +93,11 @@ func InitMirror(config *Configuration) error {
 	return configureEmptyMirror(config.MirrorNic, config.MTU)
 }
 
-func ovsInitProviderNetwork(provider, nic string, exchangeLinkName, macLearningFallback bool) (int, error) {
+func (c *Controller) ovsInitProviderNetwork(provider, nic string, exchangeLinkName, macLearningFallback bool) (int, error) {
 	// create and configure external bridge
 	brName := util.ExternalBridgeName(provider)
 	if exchangeLinkName {
-		exchanged, err := changeProvideNicName(nic, brName)
+		exchanged, err := c.changeProvideNicName(nic, brName)
 		if err != nil {
 			klog.Errorf("failed to change provider nic name from %s to %s: %v", nic, brName, err)
 			return 0, err
@@ -127,7 +123,7 @@ func ovsInitProviderNetwork(provider, nic string, exchangeLinkName, macLearningF
 
 	// add host nic to the external bridge
 	klog.Infof("config provider nic %s on bridge %s", nic, brName)
-	mtu, err := configProviderNic(nic, brName)
+	mtu, err := c.configProviderNic(nic, brName)
 	if err != nil {
 		errMsg := fmt.Errorf("failed to add nic %s to external bridge %s: %v", nic, brName, err)
 		klog.Error(errMsg)
@@ -137,7 +133,7 @@ func ovsInitProviderNetwork(provider, nic string, exchangeLinkName, macLearningF
 	return mtu, nil
 }
 
-func ovsCleanProviderNetwork(provider string) error {
+func (c *Controller) ovsCleanProviderNetwork(provider string) error {
 	mappings, err := getOvnMappings("ovn-bridge-mappings")
 	if err != nil {
 		return err
@@ -191,7 +187,7 @@ func ovsCleanProviderNetwork(provider string) error {
 	klog.V(3).Infof("ovs bridge %s has been deleted", brName)
 
 	if br := util.ExternalBridgeName(provider); br != brName {
-		if _, err = changeProvideNicName(br, brName); err != nil {
+		if _, err = c.changeProvideNicName(br, brName); err != nil {
 			klog.Errorf("failed to change provider nic name from %s to %s: %v", br, brName, err)
 			return err
 		}

pkg/daemon/init_linux.go

@@ -3,7 +3,6 @@ package daemon
 import (
 	"time"
 
-	"github.com/kubeovn/gonetworkmanager/v2"
 	"github.com/vishvananda/netlink"
 	"k8s.io/klog/v2"
 
@@ -17,46 +16,6 @@ var routeScopeOrders = [...]netlink.Scope{
 	netlink.SCOPE_UNIVERSE,
 }
 
-func nmSetManaged(device string, managed bool) error {
-	nm, err := gonetworkmanager.NewNetworkManager()
-	if err != nil {
-		klog.V(5).Infof("failed to connect to NetworkManager: %v", err)
-		return nil
-	}
-
-	running, err := nm.Running()
-	if err != nil {
-		klog.Warningf("failed to check NetworkManager running state: %v", err)
-		return nil
-	}
-	if !running {
-		klog.V(5).Info("NetworkManager is not running, ignore")
-		return nil
-	}
-
-	d, err := nm.GetDeviceByIpIface(device)
-	if err != nil {
-		klog.Errorf("failed to get device by IP iface %s: %v", device, err)
-		return err
-	}
-	current, err := d.GetPropertyManaged()
-	if err != nil {
-		klog.Errorf("failed to get device property managed: %v", err)
-		return err
-	}
-	if current == managed {
-		return nil
-	}
-
-	klog.Infof(`setting device %s NetworkManager property "managed" to %v`, device, managed)
-	if err = d.SetPropertyManaged(managed); err != nil {
-		klog.Errorf("failed to set device property managed to %v: %v", managed, err)
-		return err
-	}
-
-	return nil
-}
-
 // wait systemd-networkd to finish interface configuration
 func waitNetworkdConfiguration(linkIndex int) {
 	done := make(chan struct{})
@@ -90,7 +49,7 @@ func waitNetworkdConfiguration(linkIndex int) {
 	}
 }
 
-func changeProvideNicName(current, target string) (bool, error) {
+func (c *Controller) changeProvideNicName(current, target string) (bool, error) {
 	link, err := netlink.LinkByName(current)
 	if err != nil {
 		if _, ok := err.(netlink.LinkNotFoundError); ok {
@@ -117,8 +76,8 @@ func changeProvideNicName(current, target string) (bool, error) {
 	}
 
 	// set link unmanaged by NetworkManager
-	if err = nmSetManaged(current, false); err != nil {
-		klog.Errorf("failed set device %s unmanaged by NetworkManager: %v", current, err)
+	if err = c.nmSyncer.SetManaged(current, false); err != nil {
+		klog.Errorf("failed to set device %s unmanaged by NetworkManager: %v", current, err)
 		return false, err
 	}
 

pkg/daemon/init_windows.go

@@ -1,6 +1,6 @@
 package daemon
 
-func changeProvideNicName(nic, br string) (bool, error) {
+func (c *Controller) changeProvideNicName(nic, br string) (bool, error) {
 	// not supported on windows
 	return false, nil
 }

pkg/daemon/nm_linux.go

@@ -0,0 +1,186 @@
+package daemon
+
+import (
+	"sync"
+
+	"github.com/kubeovn/gonetworkmanager/v2"
+	"github.com/scylladb/go-set/strset"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+)
+
+type networkManagerSyncer struct {
+	manager     gonetworkmanager.NetworkManager
+	workqueue   workqueue.Interface
+	devicePaths *strset.Set
+	pathMap     map[string]string
+	bridgeMap   map[string]string
+	lock        sync.Mutex
+}
+
+func newNetworkManagerSyncer() *networkManagerSyncer {
+	syncer := &networkManagerSyncer{}
+
+	manager, err := gonetworkmanager.NewNetworkManager()
+	if err != nil {
+		klog.V(3).Infof("failed to connect to NetworkManager: %v", err)
+		return syncer
+	}
+
+	running, err := manager.Running()
+	if err != nil {
+		klog.V(3).Infof("failed to check NetworkManager running state: %v", err)
+		return syncer
+	}
+	if !running {
+		klog.V(3).Info("NetworkManager is not running, ignore")
+		return syncer
+	}
+
+	syncer.manager = manager
+	syncer.workqueue = workqueue.NewNamed("NetworkManagerSyncer")
+	syncer.devicePaths = strset.New()
+	syncer.pathMap = make(map[string]string)
+	syncer.bridgeMap = make(map[string]string)
+	return syncer
+}
+
+func (n *networkManagerSyncer) Run(handler func(nic, bridge string, delNonExistent bool) (int, error)) {
+	if n.manager == nil {
+		return
+	}
+
+	go func() {
+		for n.ProcessNextItem(handler) {
+		}
+	}()
+
+	go func() {
+		ch := n.manager.Subscribe()
+		defer n.manager.Unsubscribe()
+
+		stateChange := gonetworkmanager.DeviceInterface + "." + gonetworkmanager.ActiveConnectionSignalStateChanged
+		for {
+			event := <-ch
+
+			n.lock.Lock()
+			if len(event.Body) == 0 || event.Name != stateChange || !n.devicePaths.Has(string(event.Path)) {
+				n.lock.Unlock()
+				continue
+			}
+			n.lock.Unlock()
+
+			state, ok := event.Body[0].(uint32)
+			if !ok {
+				klog.Warningf("failed to convert %#v to uint32", event.Body[0])
+				continue
+			}
+			if gonetworkmanager.NmDeviceState(state) != gonetworkmanager.NmDeviceStateActivated {
+				continue
+			}
+
+			klog.Infof("adding dbus object path %s to workqueue", event.Path)
+			n.workqueue.Add(string(event.Path))
+		}
+	}()
+}
+
+func (n *networkManagerSyncer) ProcessNextItem(handler func(nic, bridge string, delNonExistent bool) (int, error)) bool {
+	item, shutdown := n.workqueue.Get()
+	if shutdown {
+		return false
+	}
+	defer n.workqueue.Done(item)
+
+	klog.Infof("process dbus object path %v", item)
+	path := item.(string)
+	n.lock.Lock()
+	if !n.devicePaths.Has(path) {
+		n.lock.Unlock()
+		return true
+	}
+	var nic string
+	for k, v := range n.pathMap {
+		if v == path {
+			nic = k
+			break
+		}
+	}
+	n.lock.Unlock()
+
+	bridge := n.bridgeMap[nic]
+	if _, err := handler(nic, bridge, true); err != nil {
+		klog.Errorf("failed to handle NetworkManager event for device %s with bridge %s: %v", nic, bridge, err)
+	}
+
+	return true
+}
+
+func (n *networkManagerSyncer) AddDevice(nicName, bridge string) error {
+	if n.manager == nil {
+		return nil
+	}
+
+	n.lock.Lock()
+	defer n.lock.Unlock()
+
+	if _, ok := n.pathMap[nicName]; ok {
+		return nil
+	}
+
+	device, err := n.manager.GetDeviceByIpIface(nicName)
+	if err != nil {
+		klog.Errorf("failed to get device by IP iface %q: %v", nicName, err)
+		return err
+	}
+
+	path := string(device.GetPath())
+	klog.V(3).Infof("adding device %s with dbus object path %s and bridge %s", nicName, path, bridge)
+	n.devicePaths.Add(path)
+	n.pathMap[nicName] = path
+	n.bridgeMap[nicName] = bridge
+
+	return nil
+}
+
+func (n *networkManagerSyncer) RemoveDevice(nicName string) error {
+	if n.manager == nil {
+		return nil
+	}
+
+	n.lock.Lock()
+	n.devicePaths.Remove(n.pathMap[nicName])
+	delete(n.pathMap, nicName)
+	delete(n.bridgeMap, nicName)
+	n.lock.Unlock()
+
+	return nil
+}
+
+func (n *networkManagerSyncer) SetManaged(name string, managed bool) error {
+	if n.manager == nil {
+		return nil
+	}
+
+	device, err := n.manager.GetDeviceByIpIface(name)
+	if err != nil {
+		klog.Errorf("failed to get device by IP iface %q: %v", name, err)
+		return err
+	}
+	current, err := device.GetPropertyManaged()
+	if err != nil {
+		klog.Errorf("failed to get device property managed: %v", err)
+		return err
+	}
+	if current == managed {
+		return nil
+	}
+
+	klog.Infof(`setting device %s NetworkManager property "managed" to %v`, name, managed)
+	if err = device.SetPropertyManaged(managed); err != nil {
+		klog.Errorf("failed to set device property managed to %v: %v", managed, err)
+		return err
+	}
+
+	return nil
+}