9565699003

Committed 18 Jun 2024 01:11PM UTC coverage: 52.693% (-0.06%) from 52.749%

Build # 9565699003

Build Type

Pull #4433

github

Committed by

qinqon

Commit Message

node: Use mdlayher/arp at link_network_manager

Signed-off-by: Enrique Llorente <ellorent@redhat.com>

Pull Request Pull Request #4433: node: Use mdlayher/arp at link_network_manager

Run Details

6 of 73 new or added lines in 2 files covered. (8.22%)

12 existing lines in 3 files now uncovered.

28719 of 54503 relevant lines covered (52.69%)

100.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

42.86

/go-controller/pkg/node/linkmanager/link_network_manager.go

package linkmanager

import (
        "fmt"
        "net"
        "net/netip"
        "sync"
        "time"

        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"

        "k8s.io/klog/v2"
        utilnet "k8s.io/utils/net"

        "github.com/mdlayher/arp"
        "github.com/vishvananda/netlink"
)

// Gather all suitable interface address + network mask and offer this as a service.
// Also offer address assignment to interfaces and ensure the state we want is maintained through a sync func

type LinkAddress struct {
        Link      netlink.Link
        Addresses []netlink.Addr
}

type Controller struct {
        mu              *sync.Mutex
        name            string
        ipv4Enabled     bool
        ipv6Enabled     bool
        store           map[string][]netlink.Addr
        linkHandlerFunc func(link netlink.Link) error
}

// NewController creates a controller to manage linux network interfaces
func NewController(name string, v4, v6 bool, linkHandlerFunc func(link netlink.Link) error) *Controller {
        return &Controller{
                mu:              &sync.Mutex{},
                name:            name,
                ipv4Enabled:     v4,
                ipv6Enabled:     v6,
                store:           make(map[string][]netlink.Addr),
                linkHandlerFunc: linkHandlerFunc,
        }
}

// Run starts the controller and syncs at least every syncPeriod
// linkHandlerFunc fires as an additional handler when sync runs
func (c *Controller) Run(stopCh <-chan struct{}, doneWg *sync.WaitGroup) {
        linkSubscribeOptions := netlink.LinkSubscribeOptions{
                ErrorCallback: func(err error) {
                        klog.Errorf("Failed during LinkSubscribe callback: %v", err)
                        // Note: Not calling sync() from here: it is redundant and unsafe when stopChan is closed.
                },
        }

        subscribe := func() (bool, chan netlink.LinkUpdate, error) {
                linkChan := make(chan netlink.LinkUpdate)
                if err := netlink.LinkSubscribeWithOptions(linkChan, stopCh, linkSubscribeOptions); err != nil {
                        return false, nil, err
                }
                // sync the manager with current addresses on the node
                c.sync()
                return true, linkChan, nil
        }

        c.runInternal(stopCh, doneWg, subscribe)
}

type subscribeFn func() (bool, chan netlink.LinkUpdate, error)

// runInternal can be used by testcases to provide a fake subscription function
// rather than using netlink
func (c *Controller) runInternal(stopChan <-chan struct{}, doneWg *sync.WaitGroup, subscribe subscribeFn) {

        doneWg.Add(1)
        go func() {
                defer doneWg.Done()

                linkSyncTimer := time.NewTicker(2 * time.Minute)
                defer linkSyncTimer.Stop()

                subscribed, addrChan, err := subscribe()
                if err != nil {
                        klog.Errorf("Error during netlink subscribe for Link Manager: %v", err)
                }

                for {
                        select {
                        case a, ok := <-addrChan:
                                linkSyncTimer.Reset(2 * time.Minute)
                                if !ok {
                                        if subscribed, addrChan, err = subscribe(); err != nil {
                                                klog.Errorf("Error during netlink re-subscribe due to channel closing for Link Manager: %v", err)
                                        }
                                        continue
                                }
                                if err := c.syncLinkLocked(a.Link); err != nil {
                                        klog.Errorf("Error while syncing link %q: %v", a.Link.Attrs().Name, err)
                                }

                        case <-linkSyncTimer.C:
                                if subscribed {
                                        klog.V(5).Info("Link manager calling sync() explicitly")
                                        c.sync()
                                } else {
                                        if subscribed, addrChan, err = subscribe(); err != nil {
                                                klog.Errorf("Error during netlink re-subscribe for Link Manager: %v", err)
                                        }
                                }
                        case <-stopChan:
                                return
                        }
                }
        }()

        klog.Info("Link manager is running")
}

// AddAddress stores the address in a store and ensures its applied
func (c *Controller) AddAddress(address netlink.Addr) error {
        if !c.isAddressValid(address) {
                return fmt.Errorf("address (%s) is not valid", address.String())
        }
        link, err := util.GetNetLinkOps().LinkByIndex(address.LinkIndex)
        if err != nil {
                return fmt.Errorf("no valid link associated with addresses %s: %v", address.String(), err)
        }
        klog.Infof("Link manager: adding address %s to link %s", address.String(), link.Attrs().Name)
        c.mu.Lock()
        defer c.mu.Unlock()
        c.addAddressToStore(link.Attrs().Name, address)
        return c.syncLink(link)
}

// DelAddress removes the address from the store and ensure its removed from a link
func (c *Controller) DelAddress(address netlink.Addr) error {
        if !c.isAddressValid(address) {
                return fmt.Errorf("address (%s) is not valid", address.String())
        }
        link, err := util.GetNetLinkOps().LinkByIndex(address.LinkIndex)
        if err != nil && !util.GetNetLinkOps().IsLinkNotFoundError(err) {
                return fmt.Errorf("no valid link associated with addresses %s: %v", address.String(), err)
        }
        klog.Infof("Link manager: deleting address %s from link %s", address.String(), link.Attrs().Name)
        c.mu.Lock()
        defer c.mu.Unlock()
        if err := util.GetNetLinkOps().AddrDel(link, &address); err != nil {
                if !util.GetNetLinkOps().IsLinkNotFoundError(err) {
                        return fmt.Errorf("failed to delete address %s: %v", address.String(), err)
                }
        }
        c.delAddressFromStore(link.Attrs().Name, address)
        return nil
}

// syncLinkLocked is just a wrapper around syncLink that ensures the mutex is locked in advance
func (c *Controller) syncLinkLocked(link netlink.Link) error {
        c.mu.Lock()
        defer c.mu.Unlock()
        return c.syncLink(link)
}

// syncLink handles link updates
// It MUST be called with the controller locked
func (c *Controller) syncLink(link netlink.Link) error {
        if c.linkHandlerFunc != nil {
                if err := c.linkHandlerFunc(link); err != nil {
                        klog.Errorf("Failed to execute link handler function on link: %s, error: %v", link.Attrs().Name, err)
                }
        }
        linkName := link.Attrs().Name
        // get all addresses associated with the link depending on which IP families we support
        foundAddresses, err := util.GetFilteredInterfaceAddrs(link, c.ipv4Enabled, c.ipv6Enabled)
        if err != nil {
                return fmt.Errorf("failed to get address from link %q: %w", linkName, err)
        }
        wantedAddresses, found := c.store[linkName]
        // we don't manage this link therefore we don't need to add any addresses
        if !found {
                return nil
        }
        // add addresses we want that are not found on the link
        for _, addressWanted := range wantedAddresses {
                if containsAddress(foundAddresses, addressWanted) {
                        continue
                }
                if err = util.GetNetLinkOps().AddrAdd(link, &addressWanted); err != nil {
                        klog.Errorf("Link manager: failed to add address %q to link %q: %v", addressWanted.String(), linkName, err)
                }
                // For IPv4, use arping to try to update other hosts ARP caches, in case this IP was
                // previously active on another node
                if addressWanted.IP.To4() != nil {
                        if err = gratuitousArpOverIfaceByName(addressWanted.IP, linkName); err != nil {
                                klog.Errorf("Failed to send a GARP for IP %s over interface %s: %v", addressWanted.IP.String(),
                                        linkName, err)
                        }
                }
                klog.Infof("Link manager: completed adding address %s to link %s", addressWanted, linkName)
        }
        return nil
}

func (c *Controller) sync() {
        // 1. get all the links on the node
        // 2. iterate over the links and get the addresses associated with it
        // 3. add addresses that are missing from a link that we manage
        links, err := util.GetNetLinkOps().LinkList()
        if err != nil {
                klog.Errorf("Link manager: failed to list links: %v", err)
                return
        }
        c.mu.Lock()
        defer c.mu.Unlock()
        for _, link := range links {
                if err := c.syncLink(link); err != nil {
                        klog.Errorf("Sync Link failed for link %q: %v", link.Attrs().Name, err)
                }
        }
}

func (c *Controller) addAddressToStore(linkName string, newAddress netlink.Addr) {
        addressesSaved, found := c.store[linkName]
        if !found {
                c.store[linkName] = []netlink.Addr{newAddress}
                return
        }
        // check if the address already exists
        for _, addressSaved := range addressesSaved {
                if addressSaved.Equal(newAddress) {
                        return
                }
        }
        // add it to store if not found
        c.store[linkName] = append(addressesSaved, newAddress)
}

func (c *Controller) delAddressFromStore(linkName string, address netlink.Addr) {
        addressesSaved, found := c.store[linkName]
        if !found {
                return
        }
        temp := addressesSaved[:0]
        for _, addressSaved := range addressesSaved {
                if !addressSaved.Equal(address) {
                        temp = append(temp, addressSaved)
                }
        }
        c.store[linkName] = temp
}

func (c *Controller) isAddressValid(address netlink.Addr) bool {
        if address.LinkIndex == 0 {
                return false
        }
        if address.IPNet == nil {
                return false
        }
        if address.IPNet.IP.IsUnspecified() {
                return false
        }
        if utilnet.IsIPv4(address.IP) && !c.ipv4Enabled {
                return false
        }
        if utilnet.IsIPv6(address.IP) && !c.ipv6Enabled {
                return false
        }
        return true
}

// DeprecatedGetAssignedAddressLabel returns the label that must be assigned to each egress IP address bound to an interface
func DeprecatedGetAssignedAddressLabel(linkName string) string {
        return fmt.Sprintf("%sovn", linkName)
}

func containsAddress(addresses []netlink.Addr, candidate netlink.Addr) bool {
        for _, address := range addresses {
                if address.Equal(candidate) {
                        return true
                }
        }
        return false
}

func gratuitousArpOverIfaceByName(srcIP net.IP, ifaceName string) error {
        iface, err := net.InterfaceByName(ifaceName)
        if err != nil {
                return fmt.Errorf("failed finding interface '%s': %w", ifaceName, err)
        }

        c, err := arp.Dial(iface)
        if err != nil {
                return fmt.Errorf("failed dialing logical switch interface '%s': %w", ifaceName, err)
        }
        defer c.Close()
        addr, err := netip.ParseAddr(srcIP.String())
        if err != nil {
                return fmt.Errorf("failed converting net.IP to netip.Addr: %w", err)
        }
        p, err := arp.NewPacket(arp.OperationRequest, iface.HardwareAddr, addr, net.HardwareAddr{0, 0, 0, 0, 0, 0}, addr)
        if err != nil {
                return fmt.Errorf("failed create GARP: %w", err)
        }
        err = c.WriteTo(p, net.HardwareAddr{0xff, 0xff, 0xff, 0xff, 0xff, 0xff})
        if err != nil {
                return fmt.Errorf("failed sending GARP: %w", err)
        }
        return nil
}

1	package linkmanager
2
3	import (
4	"fmt"
5	"net"
6	"net/netip"
7	"sync"
8	"time"
9
10	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
11
12	"k8s.io/klog/v2"
13	utilnet "k8s.io/utils/net"
14
15	"github.com/mdlayher/arp"
16	"github.com/vishvananda/netlink"
17	)
18
19	// Gather all suitable interface address + network mask and offer this as a service.
20	// Also offer address assignment to interfaces and ensure the state we want is maintained through a sync func
21
22	type LinkAddress struct {
23	Link netlink.Link
24	Addresses []netlink.Addr
25	}
26
27	type Controller struct {
28	mu *sync.Mutex
29	name string
30	ipv4Enabled bool
31	ipv6Enabled bool
32	store map[string][]netlink.Addr
33	linkHandlerFunc func(link netlink.Link) error
34	}
35
36	// NewController creates a controller to manage linux network interfaces
37	func NewController(name string, v4, v6 bool, linkHandlerFunc func(link netlink.Link) error) *Controller {	7✔
38	return &Controller{	7✔
39	mu: &sync.Mutex{},	7✔
40	name: name,	7✔
41	ipv4Enabled: v4,	7✔
42	ipv6Enabled: v6,	7✔
43	store: make(map[string][]netlink.Addr),	7✔
44	linkHandlerFunc: linkHandlerFunc,	7✔
45	}	7✔
46	}	7✔
47
48	// Run starts the controller and syncs at least every syncPeriod
49	// linkHandlerFunc fires as an additional handler when sync runs
50	func (c Controller) Run(stopCh <-chan struct{}, doneWg sync.WaitGroup) {	×
51	linkSubscribeOptions := netlink.LinkSubscribeOptions{	×
52	ErrorCallback: func(err error) {	×
53	klog.Errorf("Failed during LinkSubscribe callback: %v", err)	×
54	// Note: Not calling sync() from here: it is redundant and unsafe when stopChan is closed.	×
55	},	×
56	}
57
58	subscribe := func() (bool, chan netlink.LinkUpdate, error) {	×
59	linkChan := make(chan netlink.LinkUpdate)	×
60	if err := netlink.LinkSubscribeWithOptions(linkChan, stopCh, linkSubscribeOptions); err != nil {	×
61	return false, nil, err	×
62	}	×
63	// sync the manager with current addresses on the node
64	c.sync()	×
65	return true, linkChan, nil	×
66	}
67
68	c.runInternal(stopCh, doneWg, subscribe)	×
69	}
70
71	type subscribeFn func() (bool, chan netlink.LinkUpdate, error)
72
73	// runInternal can be used by testcases to provide a fake subscription function
74	// rather than using netlink
75	func (c Controller) runInternal(stopChan <-chan struct{}, doneWg sync.WaitGroup, subscribe subscribeFn) {	×
76		×
77	doneWg.Add(1)	×
78	go func() {	×
79	defer doneWg.Done()	×
80		×
81	linkSyncTimer := time.NewTicker(2 * time.Minute)	×
82	defer linkSyncTimer.Stop()	×
83		×
84	subscribed, addrChan, err := subscribe()	×
85	if err != nil {	×
86	klog.Errorf("Error during netlink subscribe for Link Manager: %v", err)	×
87	}	×
88
89	for {	×
90	select {	×
91	case a, ok := <-addrChan:	×
92	linkSyncTimer.Reset(2 * time.Minute)	×
93	if !ok {	×
94	if subscribed, addrChan, err = subscribe(); err != nil {	×
95	klog.Errorf("Error during netlink re-subscribe due to channel closing for Link Manager: %v", err)	×
96	}	×
97	continue	×
98	}
99	if err := c.syncLinkLocked(a.Link); err != nil {	×
100	klog.Errorf("Error while syncing link %q: %v", a.Link.Attrs().Name, err)	×
101	}	×
102
103	case <-linkSyncTimer.C:	×
104	if subscribed {	×
105	klog.V(5).Info("Link manager calling sync() explicitly")	×
106	c.sync()	×
107	} else {	×
108	if subscribed, addrChan, err = subscribe(); err != nil {	×
109	klog.Errorf("Error during netlink re-subscribe for Link Manager: %v", err)	×
110	}	×
111	}
112	case <-stopChan:	×
113	return	×
114	}
115	}
116	}()
117
118	klog.Info("Link manager is running")	×
119	}
120
121	// AddAddress stores the address in a store and ensures its applied
122	func (c *Controller) AddAddress(address netlink.Addr) error {	4✔
123	if !c.isAddressValid(address) {	5✔
124	return fmt.Errorf("address (%s) is not valid", address.String())	1✔
125	}	1✔
126	link, err := util.GetNetLinkOps().LinkByIndex(address.LinkIndex)	3✔
127	if err != nil {	3✔
128	return fmt.Errorf("no valid link associated with addresses %s: %v", address.String(), err)	×
129	}	×
130	klog.Infof("Link manager: adding address %s to link %s", address.String(), link.Attrs().Name)	3✔
131	c.mu.Lock()	3✔
132	defer c.mu.Unlock()	3✔
133	c.addAddressToStore(link.Attrs().Name, address)	3✔
134	return c.syncLink(link)	3✔
135	}
136
137	// DelAddress removes the address from the store and ensure its removed from a link
138	func (c *Controller) DelAddress(address netlink.Addr) error {	3✔
139	if !c.isAddressValid(address) {	4✔
140	return fmt.Errorf("address (%s) is not valid", address.String())	1✔
141	}	1✔
142	link, err := util.GetNetLinkOps().LinkByIndex(address.LinkIndex)	2✔
143	if err != nil && !util.GetNetLinkOps().IsLinkNotFoundError(err) {	2✔
144	return fmt.Errorf("no valid link associated with addresses %s: %v", address.String(), err)	×
145	}	×
146	klog.Infof("Link manager: deleting address %s from link %s", address.String(), link.Attrs().Name)	2✔
147	c.mu.Lock()	2✔
148	defer c.mu.Unlock()	2✔
149	if err := util.GetNetLinkOps().AddrDel(link, &address); err != nil {	2✔
150	if !util.GetNetLinkOps().IsLinkNotFoundError(err) {	×
151	return fmt.Errorf("failed to delete address %s: %v", address.String(), err)	×
152	}	×
153	}
154	c.delAddressFromStore(link.Attrs().Name, address)	2✔
155	return nil	2✔
156	}
157
158	// syncLinkLocked is just a wrapper around syncLink that ensures the mutex is locked in advance
159	func (c *Controller) syncLinkLocked(link netlink.Link) error {	×
160	c.mu.Lock()	×
161	defer c.mu.Unlock()	×
162	return c.syncLink(link)	×
163	}	×
164
165	// syncLink handles link updates
166	// It MUST be called with the controller locked
167	func (c *Controller) syncLink(link netlink.Link) error {	3✔
168	if c.linkHandlerFunc != nil {	3✔
169	if err := c.linkHandlerFunc(link); err != nil {	×
170	klog.Errorf("Failed to execute link handler function on link: %s, error: %v", link.Attrs().Name, err)	×
171	}	×
172	}
173	linkName := link.Attrs().Name	3✔
174	// get all addresses associated with the link depending on which IP families we support	3✔
175	foundAddresses, err := util.GetFilteredInterfaceAddrs(link, c.ipv4Enabled, c.ipv6Enabled)	3✔
176	if err != nil {	3✔
177	return fmt.Errorf("failed to get address from link %q: %w", linkName, err)	×
178	}	×
179	wantedAddresses, found := c.store[linkName]	3✔
180	// we don't manage this link therefore we don't need to add any addresses	3✔
181	if !found {	3✔
182	return nil	×
183	}	×
184	// add addresses we want that are not found on the link
185	for _, addressWanted := range wantedAddresses {	6✔
186	if containsAddress(foundAddresses, addressWanted) {	4✔
187	continue	1✔
188	}
189	if err = util.GetNetLinkOps().AddrAdd(link, &addressWanted); err != nil {	2✔
190	klog.Errorf("Link manager: failed to add address %q to link %q: %v", addressWanted.String(), linkName, err)	×
191	}	×
192	// For IPv4, use arping to try to update other hosts ARP caches, in case this IP was
193	// previously active on another node
194	if addressWanted.IP.To4() != nil {	4✔
195	if err = gratuitousArpOverIfaceByName(addressWanted.IP, linkName); err != nil {	4✔
196	klog.Errorf("Failed to send a GARP for IP %s over interface %s: %v", addressWanted.IP.String(),	2✔
197	linkName, err)	2✔
198	}	2✔
199	}
200	klog.Infof("Link manager: completed adding address %s to link %s", addressWanted, linkName)	2✔
201	}
202	return nil	3✔
203	}
204
205	func (c *Controller) sync() {	×
206	// 1. get all the links on the node	×
207	// 2. iterate over the links and get the addresses associated with it	×
208	// 3. add addresses that are missing from a link that we manage	×
209	links, err := util.GetNetLinkOps().LinkList()	×
210	if err != nil {	×
211	klog.Errorf("Link manager: failed to list links: %v", err)	×
212	return	×
213	}	×
214	c.mu.Lock()	×
215	defer c.mu.Unlock()	×
216	for _, link := range links {	×
217	if err := c.syncLink(link); err != nil {	×
218	klog.Errorf("Sync Link failed for link %q: %v", link.Attrs().Name, err)	×
219	}	×
220	}
221	}
222
223	func (c *Controller) addAddressToStore(linkName string, newAddress netlink.Addr) {	3✔
224	addressesSaved, found := c.store[linkName]	3✔
225	if !found {	4✔
226	c.store[linkName] = []netlink.Addr{newAddress}	1✔
227	return	1✔
228	}	1✔
229	// check if the address already exists
230	for _, addressSaved := range addressesSaved {	4✔
231	if addressSaved.Equal(newAddress) {	4✔
232	return	2✔
233	}	2✔
234	}
235	// add it to store if not found
236	c.store[linkName] = append(addressesSaved, newAddress)	×
237	}
238
239	func (c *Controller) delAddressFromStore(linkName string, address netlink.Addr) {	2✔
240	addressesSaved, found := c.store[linkName]	2✔
241	if !found {	2✔
242	return	×
243	}	×
244	temp := addressesSaved[:0]	2✔
245	for _, addressSaved := range addressesSaved {	4✔
246	if !addressSaved.Equal(address) {	2✔
247	temp = append(temp, addressSaved)	×
248	}	×
249	}
250	c.store[linkName] = temp	2✔
251	}
252
253	func (c *Controller) isAddressValid(address netlink.Addr) bool {	7✔
254	if address.LinkIndex == 0 {	7✔
255	return false	×
256	}	×
257	if address.IPNet == nil {	7✔
258	return false	×
259	}	×
260	if address.IPNet.IP.IsUnspecified() {	7✔
261	return false	×
262	}	×
263	if utilnet.IsIPv4(address.IP) && !c.ipv4Enabled {	9✔
264	return false	2✔
265	}	2✔
266	if utilnet.IsIPv6(address.IP) && !c.ipv6Enabled {	5✔
267	return false	×
268	}	×
269	return true	5✔
270	}
271
272	// DeprecatedGetAssignedAddressLabel returns the label that must be assigned to each egress IP address bound to an interface
273	func DeprecatedGetAssignedAddressLabel(linkName string) string {	×
274	return fmt.Sprintf("%sovn", linkName)	×
275	}	×
276
277	func containsAddress(addresses []netlink.Addr, candidate netlink.Addr) bool {	3✔
278	for _, address := range addresses {	4✔
279	if address.Equal(candidate) {	2✔
280	return true	1✔
281	}	1✔
282	}
283	return false	2✔
284	}
285
286	func gratuitousArpOverIfaceByName(srcIP net.IP, ifaceName string) error {	2✔
287	iface, err := net.InterfaceByName(ifaceName)	2✔
288	if err != nil {	4✔
289	return fmt.Errorf("failed finding interface '%s': %w", ifaceName, err)	2✔
290	}	2✔
291
NEW 292	c, err := arp.Dial(iface)	×
NEW 293	if err != nil {	×
NEW 294	return fmt.Errorf("failed dialing logical switch interface '%s': %w", ifaceName, err)	×
NEW 295	}	×
NEW 296	defer c.Close()	×
NEW 297	addr, err := netip.ParseAddr(srcIP.String())	×
NEW 298	if err != nil {	×
NEW 299	return fmt.Errorf("failed converting net.IP to netip.Addr: %w", err)	×
NEW 300	}	×
NEW 301	p, err := arp.NewPacket(arp.OperationRequest, iface.HardwareAddr, addr, net.HardwareAddr{0, 0, 0, 0, 0, 0}, addr)	×
NEW 302	if err != nil {	×
NEW 303	return fmt.Errorf("failed create GARP: %w", err)	×
NEW 304	}	×
NEW 305	err = c.WriteTo(p, net.HardwareAddr{0xff, 0xff, 0xff, 0xff, 0xff, 0xff})	×
NEW 306	if err != nil {	×
NEW 307	return fmt.Errorf("failed sending GARP: %w", err)	×
NEW 308	}	×
NEW 309	return nil	×
310	}

ovn-org / ovn-kubernetes / 9565699003

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous