fanhuadesenlinnn/kube-vip
1
0
Fork 0
mirror of https://hubproxy.babadafafafafa.cn/https://github.com/kube-vip/kube-vip.git synced 2026-09-20 08:03:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(vip): synchronize shared datapath state (#1768)

Browse Source 
* fix(vip): synchronize shared datapath state

Serialize the interface link cache, route tracker and address configuration so
concurrent Services cannot corrupt shared state or deadlock on nested address
locks.

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>

* fix: dhcp panic and leak potential

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>

---------

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>
This commit is contained in:
Marcel Fest
2026-09-18 14:29:51 +02:00
committed by GitHub
parent 3d73039cce
commit f0cbb81d93
17 changed files with 1234 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
go.mod
View File

@@ -22,7 +22,8 @@ require (
github.com/prometheus/client_golang v1.24.1
github.com/spf13/cobra v1.10.2
github.com/stretchr/testify v1.12.1
github.com/vishvananda/netlink v1.3.1
github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b
github.com/vishvananda/netns v0.0.5
go.etcd.io/etcd/api/v3 v3.7.1
go.etcd.io/etcd/client/pkg/v3 v3.7.1
go.etcd.io/etcd/client/v3 v3.7.1
@@ -120,7 +121,6 @@ require (
github.com/subosito/gotenv v1.6.0 // indirect
github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/vishvananda/netns v0.0.5 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect

6
go.sum
View File

@@ -265,8 +265,8 @@ github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701 h1:pyC9PaHYZFgEKFdlp3G8
github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701/go.mod h1:P3a5rG4X7tI17Nn3aOIAYr5HbIMukwXG0urG0WuL8OA=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/vishvananda/netlink v1.3.1 h1:3AEMt62VKqz90r0tmNhog0r/PpWKmrEShJU0wJW6bV0=
github.com/vishvananda/netlink v1.3.1/go.mod h1:ARtKouGSTGchR8aMwmkzC0qiNPrrWO5JS/XMVl45+b4=
github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b h1:XtEhFJO3IqjQWHJZ3bbNm7LtbDehriJK65KW+6lnw+Q=
github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b/go.mod h1:lEui7SPMd9fgxzHVGRAvTxsBGCF6PRH81o2kLWLWHgw=
github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zdEY=
github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
@@ -331,9 +331,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=

46
pkg/networkinterface/networkinterface.go
View File

@@ -1,19 +1,19 @@
package networkinterface
import (
log "log/slog"
"sync"
"github.com/vishvananda/netlink"
)
type Manager struct {
lock sync.Mutex
interfaces map[string]*Link
}
type Link struct {
Lock sync.Mutex
Intf netlink.Link
mu sync.Mutex
intf netlink.Link
}
func NewManager() *Manager {
@@ -23,19 +23,37 @@ func NewManager() *Manager {
}
func (m *Manager) Get(intf netlink.Link) *Link {
if l, ok := m.interfaces[intf.Attrs().Name]; ok {
updated, err := netlink.LinkByName(l.Intf.Attrs().Name)
if err != nil {
log.Error("failed to get interface %q: %w", l.Intf.Attrs().Name, err)
if intf == nil || intf.Attrs() == nil {
return nil
}
l.Intf = updated
return l
}
result := &Link{
Intf: intf,
attrs := intf.Attrs()
m.lock.Lock()
defer m.lock.Unlock()
if link, ok := m.interfaces[attrs.Name]; ok {
link.replace(intf)
return link
}
m.interfaces[intf.Attrs().Name] = result
return result
link := &Link{intf: intf}
m.interfaces[attrs.Name] = link
return link
}
func (l *Link) WithInterface(run func(netlink.Link) error) error {
l.mu.Lock()
defer l.mu.Unlock()
return run(l.intf)
}
func (l *Link) replace(intf netlink.Link) {
l.mu.Lock()
defer l.mu.Unlock()
l.intf = intf
}
func (m *Manager) Len() int {
m.lock.Lock()
defer m.lock.Unlock()
return len(m.interfaces)
}

65
pkg/networkinterface/networkinterface_instance_test.go Normal file
View File

@@ -0,0 +1,65 @@
package networkinterface_test
import (
"context"
"sync"
"testing"
"time"
"github.com/kube-vip/kube-vip/pkg/arp"
"github.com/kube-vip/kube-vip/pkg/instance"
"github.com/kube-vip/kube-vip/pkg/kubevip"
"github.com/kube-vip/kube-vip/pkg/networkinterface"
"github.com/kube-vip/kube-vip/pkg/node/noop"
"github.com/kube-vip/kube-vip/pkg/route"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
func TestManagerReconstructsProductionInstanceConcurrently(t *testing.T) {
config := &kubevip.Config{Interface: "lo", ServicesInterface: "lo", VIPSubnet: "32", DisableServiceUpdates: true}
manager := networkinterface.NewManager()
service := &v1.Service{
ObjectMeta: metav1.ObjectMeta{Name: "service", Namespace: "default", UID: "service"},
Spec: v1.ServiceSpec{LoadBalancerIP: "192.0.2.10"},
}
start := make(chan struct{})
var ready sync.WaitGroup
ready.Add(2)
results := make(chan struct {
instance *instance.Instance
err error
}, 2)
for range 2 {
go func() {
ready.Done()
<-start
instanceConfig := *config
created, err := instance.NewInstance(context.Background(), service.DeepCopy(), &instanceConfig, manager,
arp.NewManager(&instanceConfig), route.NewManager(), noop.NewManager(), &sync.WaitGroup{})
results <- struct {
instance *instance.Instance
err error
}{created, err}
}()
}
ready.Wait()
close(start)
for range 2 {
select {
case result := <-results:
if result.err != nil {
t.Fatalf("NewInstance() error = %v", result.err)
}
if len(result.instance.Clusters) != 1 {
t.Fatalf("cluster count = %d, want 1", len(result.instance.Clusters))
}
case <-time.After(3 * time.Second):
t.Fatal("timed out waiting for concurrent NewInstance calls")
}
}
if got := manager.Len(); got != 1 {
t.Fatalf("cached link count = %d, want 1", got)
}
}

74
pkg/networkinterface/networkinterface_test.go Normal file
View File

@@ -0,0 +1,74 @@
package networkinterface
import (
"sync"
"testing"
"github.com/vishvananda/netlink"
)
func TestManagerGetReplacesChangedInterfaceIndex(t *testing.T) {
manager := NewManager()
firstInterface := dummyLink("eth0", 1)
first := manager.Get(firstInterface)
if got := manager.Get(dummyLink("eth0", 1)); got != first {
t.Fatal("Get returned a new link for the same interface generation")
}
secondInterface := dummyLink("eth0", 2)
second := manager.Get(secondInterface)
if second != first {
t.Fatal("Get replaced the shared link after the interface index changed")
}
var current netlink.Link
if err := first.WithInterface(func(intf netlink.Link) error {
current = intf
return nil
}); err != nil {
t.Fatalf("WithInterface() error = %v", err)
}
if current != secondInterface {
t.Fatal("Get did not retain the new link generation")
}
}
func TestManagerGetConcurrent(t *testing.T) {
manager := NewManager()
interfaces := []netlink.Link{dummyLink("eth0", 1), dummyLink("eth1", 2)}
var wg sync.WaitGroup
results := make(chan struct {
index int
link *Link
}, 64)
for index := range cap(results) {
interfaceIndex := index % len(interfaces)
wg.Go(func() {
results <- struct {
index int
link *Link
}{index: interfaceIndex, link: manager.Get(interfaces[interfaceIndex])}
})
}
wg.Wait()
close(results)
var cached [2]*Link
for result := range results {
if result.link == nil {
t.Fatal("concurrent interface lookup returned nil")
}
if cached[result.index] == nil {
cached[result.index] = result.link
} else if result.link != cached[result.index] {
t.Fatalf("interface %d produced multiple cached Link objects", result.index)
}
}
if cached[0] == cached[1] {
t.Fatal("different interfaces shared one cached Link object")
}
}
func dummyLink(name string, index int) netlink.Link {
return &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: name, Index: index}}
}

5
pkg/route/manager.go
View File

@@ -73,6 +73,7 @@ func (m *Manager) Add(object string, r route, precheck, update bool) error {
if added {
log.Debug("[RT] added route", "path", key, "object", object)
}
m.tracker[key] = itm
}
itm.objects[object] = true
@@ -112,6 +113,8 @@ func (m *Manager) Delete(object string, r route) error {
}
func (m *Manager) Clear() {
m.mtx.Lock()
defer m.mtx.Unlock()
for _, itm := range m.tracker {
if err := itm.route.DeleteRoute(); err != nil {
log.Warn("[RT] failed to delete route", "err", err.Error())
@@ -121,6 +124,8 @@ func (m *Manager) Clear() {
}
func (m *Manager) Check(key string) bool {
m.mtx.Lock()
defer m.mtx.Unlock()
_, exists := m.tracker[key]
return exists
}

43
pkg/route/manager_test.go
View File

@@ -1,7 +1,9 @@
package route
import (
"errors"
"fmt"
"sync"
"testing"
)
@@ -184,6 +186,47 @@ func Test_MultipleRoutesAddDel(t *testing.T) {
})
}
func TestAddFailureDoesNotTrackRoute(t *testing.T) {
m := NewManager()
r := &mockRoute{hash: "failed-route", addErr: errors.New("add failed")}
if err := m.Add("service", r, false, false); err == nil {
t.Fatal("Add error = nil, want route failure")
}
if m.Check(r.RouteHash()) {
t.Fatal("failed route was tracked")
}
r.addErr = nil
if err := m.Add("service", r, false, false); err != nil {
t.Fatalf("retry Add error = %v", err)
}
if !m.Check(r.RouteHash()) {
t.Fatal("successful retry was not tracked")
}
}
func TestClearAndCheckAreSafeWithRouteUpdates(t *testing.T) {
manager := NewManager()
route := &mockRoute{hash: "concurrent-route", added: true}
if err := manager.Add("service", route, false, false); err != nil {
t.Fatalf("Add() error = %v", err)
}
var wg sync.WaitGroup
for range 20 {
wg.Go(func() {
manager.Check(route.RouteHash())
})
}
wg.Go(manager.Clear)
wg.Wait()
if manager.Check(route.RouteHash()) {
t.Fatal("route remained tracked after concurrent Clear")
}
}
type mockRoute struct {
added bool
addCalls int

18
pkg/services/services.go
View File

@@ -227,7 +227,12 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
if index == -1 {
log.Error("unable to find proper VIPConfig for the DHCPv4")
} else {
for ip := range inst.DHCPv4Client.IPChannel() {
for {
select {
case <-ctx.Done():
log.Debug("IPv4 update watcher stopping")
return
case ip := <-inst.DHCPv4Client.IPChannel():
log.Debug("IP changed", "ip", ip)
inst.VIPConfigs[index].VIP = ip
inst.DHCPInterfaceIPv4 = ip
@@ -237,7 +242,7 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
}
}
}
log.Debug("IPv4 update channel closed, stopping")
}
}
})
}
@@ -255,7 +260,12 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
if index == -1 {
log.Error("unable to find proper VIPConfig for the DHCPv6")
} else {
for ip := range inst.DHCPv6Client.IPChannel() {
for {
select {
case <-ctx.Done():
log.Debug("IPv6 update watcher stopping")
return
case ip := <-inst.DHCPv6Client.IPChannel():
log.Debug("IP changed", "ip", ip)
inst.VIPConfigs[index].VIP = ip
inst.DHCPInterfaceIPv6 = ip
@@ -265,7 +275,7 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
}
}
}
log.Debug("IPv6 update channel closed, stopping")
}
}
})
}

342
pkg/vip/address.go
View File

@@ -35,6 +35,9 @@ import (
const (
defaultValidLft = 60
iptablesComment = "%s kube-vip load balancer IP"
// Linux reserves IFA_PROTO values 0-3 for unspecified and kernel-generated
// addresses; user-space protocols start at 4. This is only a safety floor,
firstUserAddressProtocol = 4
DefaultMaskIPv4 = 32
DefaultMaskIPv6 = 128
@@ -77,8 +80,7 @@ type Network interface {
// network - This allows network configuration
type network struct {
mu sync.Mutex
mu sync.RWMutex
address *netlink.Addr
link *networkinterface.Link
ports []v1.ServicePort
@@ -168,6 +170,7 @@ func NewConfig(address string, iface string, loGlobalScope bool, subnet string,
if err != nil {
return networks, errors.Wrapf(err, "could not parse address '%s'", address)
}
markKubeVIPAddress(result.address, result.routingProtocol)
// set address as deprecated so it isn't used as source address according to RFC 3484
result.address.PreferedLft = 0
@@ -265,6 +268,7 @@ func NewConfig(address string, iface string, loGlobalScope bool, subnet string,
if result.address, err = netlink.ParseAddr(fmt.Sprintf("%s/%s", ip, s)); err != nil {
return networks, err
}
markKubeVIPAddress(result.address, result.routingProtocol)
// set ValidLft so that the VIP expires if the DNS entry is updated, otherwise it'll be refreshed by the DNS prober
result.address.ValidLft = defaultValidLft
@@ -310,7 +314,17 @@ func ListRoutesByDst(table int, dst *net.IPNet) ([]netlink.Route, error) {
return routes, nil
}
func (configurator *network) PrepareRoute() *netlink.Route {
func (configurator *network) PrepareRoute() (route *netlink.Route) {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
_ = configurator.link.WithInterface(func(intf netlink.Link) error {
route = configurator.prepareRoute(intf)
return nil
})
return route
}
func (configurator *network) prepareRoute(intf netlink.Link) *netlink.Route {
routeScope := netlink.SCOPE_UNIVERSE
if configurator.routingTableType == unix.RTN_LOCAL {
routeScope = netlink.SCOPE_LINK
@@ -318,7 +332,7 @@ func (configurator *network) PrepareRoute() *netlink.Route {
route := &netlink.Route{
Scope: routeScope,
Dst: configurator.address.IPNet,
LinkIndex: configurator.link.Intf.Attrs().Index,
LinkIndex: intf.Attrs().Index,
Table: configurator.routeTable,
Type: configurator.routingTableType,
Protocol: netlink.RouteProtocol(configurator.routingProtocol),
@@ -339,32 +353,33 @@ func NetlinkHash(r *netlink.Route) string {
}
// AddRoute - Add an IP address to a route table
func (configurator *network) AddRoute(precheck bool) (bool, error) {
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
route := configurator.PrepareRoute()
func (configurator *network) AddRoute(precheck bool) (added bool, err error) {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
err = configurator.link.WithInterface(func(intf netlink.Link) error {
route := configurator.prepareRoute(intf)
exists := false
var err error
if precheck {
exists, err = configurator.routeExists(route)
if err != nil {
return false, errors.Wrap(err, "failed to check route")
var existsErr error
exists, existsErr = configurator.routeExists(intf, route)
if existsErr != nil {
return errors.Wrap(existsErr, "failed to check route")
}
}
if !exists {
if err := netlink.RouteAdd(route); err != nil {
return false, errors.Wrap(err, "failed to add route")
if exists {
return nil
}
return true, nil
if routeErr := netlink.RouteAdd(route); routeErr != nil {
return errors.Wrap(routeErr, "failed to add route")
}
return false, nil
added = true
return nil
})
return added, err
}
func (configurator *network) routeExists(route *netlink.Route) (bool, error) {
routes, err := netlink.RouteList(configurator.link.Intf, netlink.FAMILY_ALL)
func (configurator *network) routeExists(intf netlink.Link, route *netlink.Route) (bool, error) {
routes, err := netlink.RouteList(intf, netlink.FAMILY_ALL)
if err != nil {
return false, errors.Wrap(err, "failed to list routes")
}
@@ -385,9 +400,10 @@ func (configurator *network) routeExists(route *netlink.Route) (bool, error) {
// the same-prefix interface address is processed in the same netlink batch,
// leaving the route in the kernel but never redistributed.
func (configurator *network) ReplaceRoute() error {
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
route := configurator.PrepareRoute()
configurator.mu.Lock()
defer configurator.mu.Unlock()
return configurator.link.WithInterface(func(intf netlink.Link) error {
route := configurator.prepareRoute(intf)
configurator.reassertToggle = !configurator.reassertToggle
if configurator.reassertToggle {
route.Realm = 1
@@ -395,6 +411,7 @@ func (configurator *network) ReplaceRoute() error {
route.Realm = 2
}
return netlink.RouteReplace(route)
})
}
// DeleteRoute - Delete an IP address from a route table
@@ -405,6 +422,8 @@ func (configurator *network) DeleteRoute() error {
// GetRoutes - Get an IP addresses from a route table
func (configurator *network) getRoutes() (*[]netlink.Route, error) {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
routes, err := ListRoutesByDst(configurator.routeTable, configurator.address.IPNet)
if err != nil {
return nil, fmt.Errorf("error getting routes: %w", err)
@@ -443,12 +462,22 @@ func (configurator *network) shouldSkipDAD(override bool) bool {
// precheck: if true, check if the IP already exists before adding
// skipDAD: if true, set IFA_F_NODAD flag for IPv6 addresses to skip Duplicate Address Detection
func (configurator *network) AddIP(precheck bool, skipDAD bool, minLifetime ...int) (bool, error) {
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
configurator.mu.Lock()
defer configurator.mu.Unlock()
var added bool
err := configurator.link.WithInterface(func(intf netlink.Link) error {
var addErr error
added, addErr = configurator.addIP(intf, precheck, skipDAD, minLifetime...)
return addErr
})
return added, err
}
func (configurator *network) addIP(intf netlink.Link, precheck bool, skipDAD bool, minLifetime ...int) (bool, error) {
var existing *netlink.Addr
var err error
if precheck {
if existing, err = configurator.IsSet(); err != nil {
if existing, err = configurator.isSet(intf); err != nil {
return false, errors.Wrap(err, "could not check if address exists")
}
}
@@ -479,8 +508,8 @@ func (configurator *network) AddIP(precheck bool, skipDAD bool, minLifetime ...i
}
log.Debug("replacing IP", "address", configurator.address)
if err := netlink.AddrReplace(configurator.link.Intf, configurator.address); err != nil {
return false, errors.Wrap(err, fmt.Sprintf("could not add ip to device %q", configurator.link.Intf.Attrs().Name))
if err := netlink.AddrReplace(intf, configurator.address); err != nil {
return false, errors.Wrap(err, fmt.Sprintf("could not add ip to device %q", intf.Attrs().Name))
}
if configurator.nftables {
@@ -518,7 +547,7 @@ func (configurator *network) configureNFTables() error {
"ports", configurator.ports, "service-name", configurator.serviceName)
opt := nftables.TableFamilyIPv4
if utils.IsIPv6(configurator.IP()) {
if utils.IsIPv6(configurator.address.IP.String()) {
opt = nftables.TableFamilyIPv6
}
@@ -595,12 +624,13 @@ func (configurator *network) addNftablesRulesToLimitTrafficPorts(c *nfinternal.C
}
}
firstRule, err := insertCommonNFTablesRules(c, configurator.IP(), comment)
vip := configurator.address.IP.String()
firstRule, err := insertCommonNFTablesRules(c, vip, comment)
if err != nil {
return fmt.Errorf("could not add common nftables rules: %w", err)
}
if err := configurator.insertNFTablesRulesForServicePorts(c, configurator.IP(), comment, firstRule.Handle); err != nil {
if err := configurator.insertNFTablesRulesForServicePorts(c, vip, comment, firstRule.Handle); err != nil {
return fmt.Errorf("could not add nftables rules for service ports: %v", err)
}
@@ -1057,10 +1087,19 @@ func (configurator *network) removeNftablesRuleToLimitTrafficPorts(c *nfinternal
// DeleteIP - Remove an IP address from the interface
func (configurator *network) DeleteIP() (bool, error) {
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
configurator.mu.Lock()
defer configurator.mu.Unlock()
var deleted bool
err := configurator.link.WithInterface(func(intf netlink.Link) error {
var deleteErr error
deleted, deleteErr = configurator.deleteIP(intf)
return deleteErr
})
return deleted, err
}
result, err := configurator.IsSet()
func (configurator *network) deleteIP(intf netlink.Link) (bool, error) {
result, err := configurator.isSet(intf)
if err != nil {
return false, errors.Wrap(err, "ip check in DeleteIP failed")
}
@@ -1070,7 +1109,7 @@ func (configurator *network) DeleteIP() (bool, error) {
return false, nil
}
if err = netlink.AddrDel(configurator.link.Intf, configurator.address); err != nil {
if err = netlink.AddrDel(intf, configurator.address); err != nil {
return false, errors.Wrap(err, "could not delete ip")
}
@@ -1140,7 +1179,8 @@ func (configurator *network) addIptablesRulesForMasquerade() error {
// TO DO: It seems it is not be possible to use google/nftables with IPVS due to lack of IPVS matcher in nft
func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client, comment string) error {
cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.IP(), configurator.IPVSMark())
vip := configurator.address.IP.String()
cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, vip, configurator.ipvsMark)
markChain := &nftables.Chain{
Name: "ipvs_prerouting",
@@ -1154,7 +1194,7 @@ func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client,
markChain = c.AddChain(markChain)
ip := net.ParseIP(configurator.IP())
ip := net.ParseIP(vip)
if ip.To4() != nil {
ip = ip.To4()
@@ -1164,7 +1204,7 @@ func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client,
port := binaryutil.BigEndian.PutUint16(configurator.ipvsPort)
mark := binaryutil.NativeEndian.PutUint32(configurator.IPVSMark())
mark := binaryutil.NativeEndian.PutUint32(configurator.ipvsMark)
markRule := &nftables.Rule{
Table: markChain.Table,
@@ -1286,7 +1326,7 @@ func (configurator *network) removeNftablesRulesForMasquerade(c *nfinternal.Clie
if configurator.serviceName != "" {
comment = fmt.Sprintf(iptablesComment, configurator.serviceName)
}
cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.IP(), configurator.IPVSMark())
cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.address.IP.String(), configurator.ipvsMark)
r, err := c.FindRuleByComment(chain.Table, chain, cmt)
if err != nil {
@@ -1349,27 +1389,26 @@ func delMasqueradeRuleForVIP(ipt *iptables.IPTables, vip, comment string) error
// IsDADFAIL - Returns true if the address is IPv6 and has DADFAILED flag
func (configurator *network) IsDADFAIL() bool {
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
configurator.mu.RLock()
defer configurator.mu.RUnlock()
var dadFailed bool
_ = configurator.link.WithInterface(func(intf netlink.Link) error {
if configurator.address == nil || !utils.IsIPv6(configurator.address.IP.String()) {
return false
return nil
}
// Get all the address
addresses, err := netlink.AddrList(configurator.link.Intf, netlink.FAMILY_V6)
addresses, err := netlink.AddrList(intf, netlink.FAMILY_V6)
if err != nil {
return false
return nil
}
// Find the VIP and check if it is DADFAILED
for _, address := range addresses {
if address.IP.Equal(configurator.address.IP) && addressHasDADFAILEDFlag(address) {
return true
dadFailed = true
break
}
}
return false
return nil
})
return dadFailed
}
func addressHasDADFAILEDFlag(address netlink.Addr) bool {
@@ -1378,6 +1417,16 @@ func addressHasDADFAILEDFlag(address netlink.Addr) bool {
// isSet - Check to see if VIP is set
func (configurator *network) IsSet() (result *netlink.Addr, err error) {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
err = configurator.link.WithInterface(func(intf netlink.Link) error {
result, err = configurator.isSet(intf)
return err
})
return result, err
}
func (configurator *network) isSet(intf netlink.Link) (result *netlink.Addr, err error) {
var addresses []netlink.Addr
if configurator.address == nil {
@@ -1388,7 +1437,7 @@ func (configurator *network) IsSet() (result *netlink.Addr, err error) {
return nil, nil
}
addresses, err = netlink.AddrList(configurator.link.Intf, 0)
addresses, err = netlink.AddrList(intf, 0)
if err != nil {
err = errors.Wrap(err, "could not list addresses")
@@ -1408,10 +1457,10 @@ func (configurator *network) IsSet() (result *netlink.Addr, err error) {
func (configurator *network) SetIP(ip string) error {
configurator.mu.Lock()
defer configurator.mu.Unlock()
return configurator.setIP(ip)
}
configurator.link.Lock.Lock()
defer configurator.link.Lock.Unlock()
func (configurator *network) setIP(ip string) error {
if strings.Contains("/", ip) {
return fmt.Errorf("ip should not contain CIDR notation got: %s", ip)
}
@@ -1443,7 +1492,7 @@ func (configurator *network) SetIP(ip string) error {
if err != nil {
return err
}
if configurator.address != nil && configurator.IsDNS() {
if configurator.address != nil && configurator.dnsName != "" {
addr.ValidLft = defaultValidLft
} else {
addr.ValidLft = math.MaxInt
@@ -1453,9 +1502,94 @@ func (configurator *network) SetIP(ip string) error {
addr.PreferedLft = 0
configurator.address = addr
if configurator.routingProtocol != 0 {
markKubeVIPAddress(configurator.address, configurator.routingProtocol)
}
return nil
}
// IsKubeVIPAddress reports whether an address carries kube-vip's configured
// protocol origin. IFA_PROTO is supported for both IPv4 and IPv6 on Linux 5.18+.
func IsKubeVIPAddress(address netlink.Addr, protocol int) bool {
return protocol >= firstUserAddressProtocol && address.Protocol == protocol
}
// RetainedKubeVIPAddressKeys returns the tagged kernel addresses matching the
// supplied VIPs. Callers use the keys with CleanupKubeVIPAddresses after they
// determine which VIPs remain locally referenced.
func RetainedKubeVIPAddressKeys(protocol int, vips map[string]struct{}) (map[string]struct{}, error) {
retained := make(map[string]struct{})
err := forEachKubeVIPAddress(protocol, func(_ netlink.Link, address netlink.Addr) error {
if _, retain := vips[address.IP.String()]; retain {
retained[addressKey(address)] = struct{}{}
}
return nil
})
if err != nil {
return nil, err
}
return retained, nil
}
// CleanupKubeVIPAddresses removes kube-vip addresses not reasserted by this
// process. The retained keys must come from successful AddrReplace operations.
func CleanupKubeVIPAddresses(protocol int, retained map[string]struct{}) (int, error) {
removed := 0
err := forEachKubeVIPAddress(protocol, func(link netlink.Link, address netlink.Addr) error {
key := addressKey(address)
if key == "" {
return nil
}
if _, keep := retained[key]; keep {
return nil
}
if err := netlink.AddrDel(link, &address); err != nil {
return errors.Wrapf(err, "deleting kube-vip address %q from interface %q", address.IP, link.Attrs().Name)
}
removed++
return nil
})
if err != nil {
return removed, err
}
return removed, nil
}
func forEachKubeVIPAddress(protocol int, visit func(netlink.Link, netlink.Addr) error) error {
links, err := netlink.LinkList()
if err != nil {
return errors.Wrap(err, "listing network links")
}
for _, link := range links {
addresses, err := netlink.AddrList(link, netlink.FAMILY_ALL)
if err != nil {
return errors.Wrapf(err, "listing addresses on interface %q", link.Attrs().Name)
}
for _, address := range addresses {
if IsKubeVIPAddress(address, protocol) {
if err := visit(link, address); err != nil {
return err
}
}
}
}
return nil
}
func addressKey(address netlink.Addr) string {
if address.LinkIndex <= 0 || address.IP == nil {
return ""
}
prefixLength, _ := address.Mask.Size()
return fmt.Sprintf("%d/%s/%d", address.LinkIndex, address.IP, prefixLength)
}
func markKubeVIPAddress(address *netlink.Addr, protocol int) {
if address != nil {
address.Protocol = protocol
}
}
// SetServicePorts updates the service ports from the service
// If you want to limit traffic to the VIP to only the service ports, add service ports to the network firstly.
func (configurator *network) SetServicePorts(service *v1.Service) {
@@ -1469,8 +1603,8 @@ func (configurator *network) SetServicePorts(service *v1.Service) {
// IP - return the IP Address
func (configurator *network) IP() string {
configurator.mu.Lock()
defer configurator.mu.Unlock()
configurator.mu.RLock()
defer configurator.mu.RUnlock()
if configurator.address == nil || configurator.address.IP == nil {
return ""
@@ -1480,8 +1614,8 @@ func (configurator *network) IP() string {
}
func (configurator *network) CIDR() string {
configurator.mu.Lock()
defer configurator.mu.Unlock()
configurator.mu.RLock()
defer configurator.mu.RUnlock()
if configurator.address == nil || configurator.address.IPNet == nil {
return ""
@@ -1492,24 +1626,30 @@ func (configurator *network) CIDR() string {
// IP - return the IP Address
func (configurator *network) IPisLinkLocal() bool {
configurator.mu.Lock()
defer configurator.mu.Unlock()
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.address.IP.IsLinkLocalUnicast()
}
// DNSName return the configured dnsName when use DNS
func (configurator *network) DNSName() string {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.dnsName
}
// IsDNS - when dnsName is configured
func (configurator *network) IsDNS() bool {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.dnsName != ""
}
// IsDDNS - return true if use dynamic dns
func (configurator *network) IsDDNS() bool {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.isDDNS
}
@@ -1518,12 +1658,19 @@ func (configurator *network) IsDDNS() bool {
// it's expected that dynamic DNS should be configured so
// the fqdn for apiserver endpoint is dDNSHostName.{LocalDomain}
func (configurator *network) DDNSHostName() string {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return getHostName(configurator.dnsName)
}
// Interface - return the Interface name
func (configurator *network) Interface() string {
return configurator.link.Intf.Attrs().Name
var name string
_ = configurator.link.WithInterface(func(intf netlink.Link) error {
name = intf.Attrs().Name
return nil
})
return name
}
func GarbageCollect(adapter, address string, intfMgr *networkinterface.Manager) (found bool, err error) {
@@ -1534,32 +1681,28 @@ func GarbageCollect(adapter, address string, intfMgr *networkinterface.Manager)
}
l := intfMgr.Get(link)
l.Lock.Lock()
defer l.Lock.Unlock()
// Get addresses on adapter
addrs, err := netlink.AddrList(l.Intf, netlink.FAMILY_ALL)
if err != nil {
return false, err
err = l.WithInterface(func(intf netlink.Link) error {
addrs, listErr := netlink.AddrList(intf, netlink.FAMILY_ALL)
if listErr != nil {
return listErr
}
// Compare all addresses to new service address, and remove if needed
for _, existing := range addrs {
if existing.IP.String() == address {
// We've found the existing address
found = true
// linting issue
existing := existing
if err = netlink.AddrDel(l.Intf, &existing); err != nil {
return true, errors.Wrap(err, "could not delete ip")
if deleteErr := netlink.AddrDel(intf, &existing); deleteErr != nil {
return errors.Wrap(deleteErr, "could not delete ip")
}
}
}
return // Didn't find the address on the adapter
return nil
})
return found, err
}
func (configurator *network) SetMask(mask string) error {
configurator.mu.Lock()
defer configurator.mu.Unlock()
selectedMask := mask
var err error
@@ -1567,8 +1710,12 @@ func (configurator *network) SetMask(mask string) error {
return fmt.Errorf("no mask provided")
}
if configurator.IP() != "" {
selectedMask, err = SelectSubnet(configurator.IP(), mask)
ip := ""
if configurator.address != nil && configurator.address.IP != nil {
ip = configurator.address.IP.String()
}
if ip != "" {
selectedMask, err = SelectSubnet(ip, mask)
if err != nil {
return fmt.Errorf("failed to select mask %q: %w", mask, err)
}
@@ -1584,8 +1731,8 @@ func (configurator *network) SetMask(mask string) error {
size := DefaultMaskIPv4
family := utils.IPv4Family
if configurator.IP() != "" {
if utils.IsIPv6(configurator.IP()) {
if ip != "" {
if utils.IsIPv6(ip) {
size = DefaultMaskIPv6
family = utils.IPv6Family
}
@@ -1600,20 +1747,29 @@ func (configurator *network) SetMask(mask string) error {
return fmt.Errorf("failed to create mask /%d", m)
}
configurator.mu.Lock()
defer configurator.mu.Unlock()
configurator.address.Mask = toSet
return nil
}
func (configurator *network) SetHasEndpoints(value bool) {
log.Debug("setting HasEndpoints", "ip", configurator.IP(), "value", value)
configurator.mu.Lock()
defer configurator.mu.Unlock()
ip := ""
if configurator.address != nil && configurator.address.IP != nil {
ip = configurator.address.IP.String()
}
log.Debug("setting HasEndpoints", "ip", ip, "value", value)
configurator.hasEndpoints = value
}
func (configurator *network) HasEndpoints() bool {
log.Debug("getting HasEndpoints", "ip", configurator.IP(), "value", configurator.hasEndpoints)
configurator.mu.RLock()
defer configurator.mu.RUnlock()
ip := ""
if configurator.address != nil && configurator.address.IP != nil {
ip = configurator.address.IP.String()
}
log.Debug("getting HasEndpoints", "ip", ip, "value", configurator.hasEndpoints)
return configurator.hasEndpoints
}
@@ -1622,14 +1778,20 @@ func (configurator *network) ARPName() string {
}
func (configurator *network) GetPossibleSubnets() string {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.possibleSubnets
}
func (configurator *network) DHCPFamily() string {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.dhcpFamily
}
func (configurator *network) IPVSMark() uint32 {
configurator.mu.RLock()
defer configurator.mu.RUnlock()
return configurator.ipvsMark
}

5
pkg/vip/address_dad_linux_test.go
View File

@@ -16,11 +16,10 @@ func TestAddIPPerCallDADSkipDoesNotPersist(t *testing.T) {
t.Fatal(err)
}
interfaceManager := networkinterface.NewManager()
configurator := &network{
address: address,
link: &networkinterface.Link{
Intf: &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kube-vip-dad-test"}},
},
link: interfaceManager.Get(&netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kube-vip-dad-test"}}),
}
// The netlink operation may fail without CAP_NET_ADMIN, but the address

220
pkg/vip/address_label_linux_test.go Normal file
View File

@@ -0,0 +1,220 @@
//go:build linux
package vip
import (
"net"
"os"
"runtime"
"testing"
"github.com/vishvananda/netlink"
"github.com/vishvananda/netns"
)
const kubeVIPProtocol = 248
// requireNetworkNamespaces makes the privileged CI job fail instead of silently
// skipping when it cannot enter a network namespace.
var requireNetworkNamespaces = os.Getenv("KUBE_VIP_REQUIRE_NETNS") != ""
func TestAddressProtocolRoundTripsThroughNetlink(t *testing.T) {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
originalNamespace, err := netns.Get()
if err != nil {
t.Fatalf("getting current network namespace: %v", err)
}
defer originalNamespace.Close()
testNamespace, err := netns.New()
if err != nil {
if requireNetworkNamespaces {
t.Fatalf("creating isolated network namespace: %v", err)
}
t.Skipf("creating isolated network namespace: %v", err)
}
defer testNamespace.Close()
defer func() {
if err := netns.Set(originalNamespace); err != nil {
t.Errorf("restoring network namespace: %v", err)
}
}()
link := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kvproto0"}}
if err := netlink.LinkAdd(link); err != nil {
t.Fatalf("creating test interface: %v", err)
}
if err := netlink.LinkSetUp(link); err != nil {
t.Fatalf("bringing test interface up: %v", err)
}
parsed, err := netlink.ParseAddr("192.0.2.10/32")
if err != nil {
t.Fatalf("parsing IPv4 address: %v", err)
}
markKubeVIPAddress(parsed, kubeVIPProtocol)
if err := netlink.AddrReplace(link, parsed); err != nil {
t.Fatalf("adding IPv4 address with protocol: %v", err)
}
addresses, err := netlink.AddrList(link, netlink.FAMILY_ALL)
if err != nil {
t.Fatalf("listing addresses: %v", err)
}
for _, configured := range addresses {
if configured.IP.Equal(net.ParseIP("192.0.2.10")) {
if !IsKubeVIPAddress(configured, kubeVIPProtocol) {
t.Fatalf("configured address = %+v, want kube-vip protocol", configured)
}
return
}
}
t.Fatal("IPv4 address with kube-vip protocol was not configured")
}
func TestKubeVIPAddressProtocolRoundTripsThroughIPv6Netlink(t *testing.T) {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
originalNamespace, err := netns.Get()
if err != nil {
t.Fatalf("getting current network namespace: %v", err)
}
defer originalNamespace.Close()
testNamespace, err := netns.New()
if err != nil {
if requireNetworkNamespaces {
t.Fatalf("creating isolated network namespace: %v", err)
}
t.Skipf("creating isolated network namespace: %v", err)
}
defer testNamespace.Close()
defer func() {
if err := netns.Set(originalNamespace); err != nil {
t.Errorf("restoring network namespace: %v", err)
}
}()
link := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kvproto1"}}
if err := netlink.LinkAdd(link); err != nil {
t.Fatalf("creating test interface: %v", err)
}
if err := netlink.LinkSetUp(link); err != nil {
t.Fatalf("bringing test interface up: %v", err)
}
address, err := netlink.ParseAddr("2001:db8::10/128")
if err != nil {
t.Fatalf("parsing IPv6 address: %v", err)
}
markKubeVIPAddress(address, kubeVIPProtocol)
if err := netlink.AddrReplace(link, address); err != nil {
t.Fatalf("adding IPv6 address with protocol: %v", err)
}
addresses, err := netlink.AddrList(link, netlink.FAMILY_V6)
if err != nil {
t.Fatalf("listing IPv6 addresses: %v", err)
}
for _, configured := range addresses {
if configured.IP.Equal(net.ParseIP("2001:db8::10")) {
if !IsKubeVIPAddress(configured, kubeVIPProtocol) {
t.Fatalf("configured IPv6 address = %+v, want kube-vip protocol", configured)
}
return
}
}
t.Fatalf("configured IPv6 addresses = %+v, want 2001:db8::10", addresses)
}
func TestCleanupKubeVIPAddressesRemovesOnlyUnretainedProtocolAddresses(t *testing.T) {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
originalNamespace, err := netns.Get()
if err != nil {
t.Fatalf("getting current network namespace: %v", err)
}
defer originalNamespace.Close()
testNamespace, err := netns.New()
if err != nil {
if requireNetworkNamespaces {
t.Fatalf("creating isolated network namespace: %v", err)
}
t.Skipf("creating isolated network namespace: %v", err)
}
defer testNamespace.Close()
defer func() {
if err := netns.Set(originalNamespace); err != nil {
t.Errorf("restoring network namespace: %v", err)
}
}()
link := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kvproto2"}}
if err := netlink.LinkAdd(link); err != nil {
t.Fatalf("creating test interface: %v", err)
}
if err := netlink.LinkSetUp(link); err != nil {
t.Fatalf("bringing test interface up: %v", err)
}
for _, input := range []struct {
cidr string
protocol int
}{
{cidr: "192.0.2.10/32", protocol: kubeVIPProtocol},
{cidr: "192.0.2.11/32", protocol: kubeVIPProtocol},
{cidr: "192.0.2.12/32", protocol: 0},
{cidr: "192.0.2.13/32", protocol: kubeVIPProtocol + 1},
{cidr: "2001:db8::10/128", protocol: kubeVIPProtocol},
{cidr: "2001:db8::11/128", protocol: kubeVIPProtocol},
} {
address, err := netlink.ParseAddr(input.cidr)
if err != nil {
t.Fatalf("parsing address: %v", err)
}
markKubeVIPAddress(address, input.protocol)
if err := netlink.AddrReplace(link, address); err != nil {
t.Fatalf("adding address %s: %v", input.cidr, err)
}
}
retained, err := netlink.ParseAddr("192.0.2.10/32")
if err != nil {
t.Fatalf("parsing retained address: %v", err)
}
retained.LinkIndex = link.Attrs().Index
retainedIPv6, err := netlink.ParseAddr("2001:db8::10/128")
if err != nil {
t.Fatalf("parsing retained IPv6 address: %v", err)
}
retainedIPv6.LinkIndex = link.Attrs().Index
removed, err := CleanupKubeVIPAddresses(kubeVIPProtocol, map[string]struct{}{
addressKey(*retained): {},
addressKey(*retainedIPv6): {},
})
if err != nil {
t.Fatalf("cleaning kube-vip addresses: %v", err)
}
if removed != 2 {
t.Fatalf("removed = %d, want 2", removed)
}
addresses, err := netlink.AddrList(link, netlink.FAMILY_ALL)
if err != nil {
t.Fatalf("listing remaining addresses: %v", err)
}
wantAddresses := map[string]bool{
"192.0.2.10": false,
"192.0.2.12": false,
"192.0.2.13": false,
"2001:db8::10": false,
}
for _, address := range addresses {
if _, wanted := wantAddresses[address.IP.String()]; wanted {
wantAddresses[address.IP.String()] = true
}
}
for address, found := range wantAddresses {
if !found {
t.Fatalf("remaining addresses = %+v, missing %s", addresses, address)
}
}
}

24
pkg/vip/address_test.go
View File

@@ -2,8 +2,32 @@ package vip
import (
"testing"
"github.com/kube-vip/kube-vip/pkg/networkinterface"
)
func TestNewConfigTagsStaticAndUpdatedAddresses(t *testing.T) {
const protocol = 248
networks, err := NewConfig("192.0.2.10", "lo", false, "32", false, "", false, false, 0, 0, protocol,
"", "", "", false, 0, false, networkinterface.NewManager(), false, false)
if err != nil {
t.Fatalf("NewConfig() error = %v", err)
}
configured, ok := networks[0].(*network)
if !ok {
t.Fatalf("network type = %T, want *network", networks[0])
}
if configured.address.Protocol != protocol {
t.Fatalf("static address protocol = %d, want %d", configured.address.Protocol, protocol)
}
if err := configured.SetIP("192.0.2.11"); err != nil {
t.Fatalf("SetIP() error = %v", err)
}
if configured.address.Protocol != protocol {
t.Fatalf("updated address protocol = %d, want %d", configured.address.Protocol, protocol)
}
}
func TestShouldSkipDAD(t *testing.T) {
cases := []struct {
name string

8
pkg/vip/dhcpv4.go
View File

@@ -73,7 +73,6 @@ func (c *DHCPv4Client) Stop() {
func (c *DHCPv4Client) close() {
c.stopOnce.Do(func() {
close(c.ipChan)
close(c.stopChan)
})
}
@@ -284,7 +283,12 @@ RequestLoop:
}
if c.ipChan != nil {
c.ipChan <- lease.ACK.YourIPAddr.String()
// Nothing closes ipChan, so never block on a consumer that already stopped.
select {
case c.ipChan <- lease.ACK.YourIPAddr.String():
case <-c.stopChan:
case <-ctx.Done():
}
}
return lease, nil

38
pkg/vip/dhcpv6.go
View File

@@ -23,6 +23,9 @@ func init() {
}
type DHCPv6ClientManager struct {
// mu guards clients and the reference counts of its entries together, so a
// concurrent Add cannot join a client that Delete is already retiring.
mu sync.Mutex
clients map[string]*DHCPv6InternalClient
}
@@ -33,17 +36,17 @@ func NewDHCPv6ClientManager() *DHCPv6ClientManager {
}
func (m *DHCPv6ClientManager) Get(iface string) *DHCPv6InternalClient {
c, exists := m.clients[iface]
if !exists {
return nil
}
return c
m.mu.Lock()
defer m.mu.Unlock()
return m.clients[iface]
}
func (m *DHCPv6ClientManager) Add(iface string) (*DHCPv6InternalClient, error) {
c := m.Get(iface)
m.mu.Lock()
defer m.mu.Unlock()
if c != nil {
if c := m.clients[iface]; c != nil {
c.references.Add(1)
return c, nil
}
@@ -57,16 +60,17 @@ func (m *DHCPv6ClientManager) Add(iface string) (*DHCPv6InternalClient, error) {
}
func (m *DHCPv6ClientManager) Delete(iface string) {
c := m.Get(iface)
m.mu.Lock()
defer m.mu.Unlock()
if c != nil {
c.references.Add(-1)
ref := c.references.Load()
if ref < 1 {
c := m.clients[iface]
if c == nil {
return
}
if c.references.Add(-1) < 1 {
c.client.Close()
delete(m.clients, iface)
}
}
}
type DHCPv6InternalClient struct {
@@ -154,7 +158,6 @@ func (c *DHCPv6Client) Stop() {
// Close dhcp client channels
func (c *DHCPv6Client) close() {
c.stop.Do(func() {
close(c.ipChan)
close(c.stopChan)
})
dhcpv6ClientManager.Delete(c.managerKey)
@@ -304,7 +307,12 @@ RequestLoop:
}
if c.ipChan != nil {
c.ipChan <- addr.IPv6Addr.String()
// Nothing closes ipChan, so never block on a consumer that already stopped.
select {
case c.ipChan <- addr.IPv6Addr.String():
case <-c.stopChan:
case <-ctx.Done():
}
}
return addr, nil

33
pkg/vip/dhcpv6_robustness_test.go
View File

@@ -2,6 +2,7 @@ package vip
import (
"net"
"sync"
"sync/atomic"
"testing"
@@ -36,6 +37,38 @@ func TestDHCPv6StopReleasesManagerReferenceForParentInterface(t *testing.T) {
}
}
func TestDHCPv6ClientManagerSharesOneClientPerParentInterface(t *testing.T) {
references := &atomic.Int32{}
references.Store(1)
shared := &DHCPv6InternalClient{references: references}
manager := &DHCPv6ClientManager{
clients: map[string]*DHCPv6InternalClient{"parent0": shared},
}
var wg sync.WaitGroup
for range 64 {
wg.Go(func() {
client, err := manager.Add("parent0")
if err != nil {
t.Errorf("Add() error = %v", err)
return
}
if client != shared {
t.Errorf("Add() client = %p, want the shared client %p", client, shared)
}
manager.Delete("parent0")
})
}
wg.Wait()
if got := manager.Get("parent0"); got != shared {
t.Fatalf("shared client = %v, want it retained while still referenced", got)
}
if got := references.Load(); got != 1 {
t.Fatalf("manager reference count = %d, want 1", got)
}
}
func TestGetAddressRejectsIANAWithoutAddresses(t *testing.T) {
// DEFECT: getAddress indexes the first IAADDR without checking whether the IANA contains one, so a malformed/expired reply panics (pkg/vip/dhcpv6.go:392).
defer func() {

110
pkg/vip/util.go
View File

@@ -7,6 +7,7 @@ import (
"net"
"strings"
"syscall"
"time"
log "log/slog"
@@ -14,6 +15,8 @@ import (
"github.com/vishvananda/netlink"
)
var errDefaultInterfaceSubscriptionClosed = errors.New("default interface subscription closed")
// getHostName return the hostname from the fqdn
func getHostName(dnsName string) string {
if dnsName == "" {
@@ -75,26 +78,123 @@ func getDefaultRoute(family int) (*net.Interface, error) {
return nil, errors.New("default route not found")
}
// MonitorDefaultInterface monitor the default interface and catch the event of the default route
// MonitorDefaultInterface monitors the default interface for route removal or link loss.
func MonitorDefaultInterface(ctx context.Context, defaultIF *net.Interface) error {
routeCh := make(chan netlink.RouteUpdate)
return monitorDefaultInterfaceWithRetry(ctx, defaultIF, subscribeDefaultInterface, GetDefaultGatewayInterface, 100*time.Millisecond)
}
func monitorDefaultInterfaceWithRetry(ctx context.Context, defaultIF *net.Interface,
subscribe func(context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error),
lookup func() (*net.Interface, error), retryDelay time.Duration) error {
for {
monitorCtx, cancel := context.WithCancel(ctx)
routeCh, linkCh, err := subscribe(monitorCtx)
if err == nil {
err = monitorDefaultInterface(monitorCtx, defaultIF, routeCh, linkCh)
}
cancel()
drainDefaultInterfaceSubscriptions(routeCh, linkCh)
if ctx.Err() != nil {
return nil
}
if err != nil && !errors.Is(err, errDefaultInterfaceSubscriptionClosed) {
log.Warn("default interface subscription failed, retrying", "err", err)
} else if err == nil {
return nil
}
if refreshed, lookupErr := lookup(); lookupErr == nil {
defaultIF = refreshed
} else {
log.Warn("failed to refresh default interface while resubscribing", "err", lookupErr)
}
timer := time.NewTimer(retryDelay)
select {
case <-ctx.Done():
timer.Stop()
return nil
case <-timer.C:
}
}
}
func subscribeDefaultInterface(ctx context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error) {
const subscriptionBuffer = 64
routeCh := make(chan netlink.RouteUpdate, subscriptionBuffer)
if err := netlink.RouteSubscribe(routeCh, ctx.Done()); err != nil {
return fmt.Errorf("subscribe route failed, error: %w", err)
return nil, nil, fmt.Errorf("subscribe route failed, error: %w", err)
}
linkCh := make(chan netlink.LinkUpdate, subscriptionBuffer)
if err := netlink.LinkSubscribe(linkCh, ctx.Done()); err != nil {
return routeCh, nil, fmt.Errorf("subscribe link failed, error: %w", err)
}
return routeCh, linkCh, nil
}
func monitorDefaultInterface(ctx context.Context, defaultIF *net.Interface, routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) error {
for {
select {
case r := <-routeCh:
case r, ok := <-routeCh:
if !ok {
return subscriptionClosed(ctx, "route")
}
log.Debug(fmt.Sprintf("type: %d, route: %+v", r.Type, r.Route))
if r.Type == syscall.RTM_DELROUTE && (r.Dst == nil || r.Dst.String() == "0.0.0.0/0") && r.LinkIndex == defaultIF.Index {
if r.Type == syscall.RTM_DELROUTE && isDefaultRoute(r.Dst) && r.LinkIndex == defaultIF.Index {
return fmt.Errorf("default route deleted and the default interface may be invalid")
}
case update, ok := <-linkCh:
if !ok {
return subscriptionClosed(ctx, "link")
}
if update.Link == nil {
continue
}
attrs := update.Attrs()
if attrs != nil && attrs.Index == defaultIF.Index && attrs.Flags&net.FlagUp == 0 {
return fmt.Errorf("default interface %q is down", defaultIF.Name)
}
case <-ctx.Done():
return nil
}
}
}
func subscriptionClosed(ctx context.Context, subscription string) error {
if ctx.Err() != nil {
return nil
}
return fmt.Errorf("%w: %s subscription closed", errDefaultInterfaceSubscriptionClosed, subscription)
}
// isDefaultRoute accepts both families, matching the selection made by
// GetDefaultGatewayInterface.
func isDefaultRoute(dst *net.IPNet) bool {
if dst == nil {
return true
}
return dst.String() == "0.0.0.0/0" || dst.String() == "::/0"
}
func drainDefaultInterfaceSubscriptions(routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) {
timer := time.NewTimer(100 * time.Millisecond)
defer timer.Stop()
for routeCh != nil || linkCh != nil {
select {
case _, ok := <-routeCh:
if !ok {
routeCh = nil
}
case _, ok := <-linkCh:
if !ok {
linkCh = nil
}
case <-timer.C:
return
}
}
}
func GenerateMac() (mac string) {
buf := make([]byte, 3)
_, err := rand.Read(buf)

293
pkg/vip/util_linux_test.go Normal file
View File

@@ -0,0 +1,293 @@
//go:build linux
package vip
import (
"context"
"net"
"runtime"
"strings"
"syscall"
"testing"
"time"
"github.com/vishvananda/netlink"
"github.com/vishvananda/netns"
)
func TestMonitorDefaultInterfaceDetectsDefaultRouteDeletionPerFamily(t *testing.T) {
for _, test := range []struct {
name string
cidr string
}{
{name: "IPv4", cidr: "0.0.0.0/0"},
{name: "IPv6", cidr: "::/0"},
} {
t.Run(test.name, func(t *testing.T) {
defaultIF := &net.Interface{Index: 7, Name: "test0"}
_, defaultRoute, err := net.ParseCIDR(test.cidr)
if err != nil {
t.Fatalf("ParseCIDR() error = %v", err)
}
routeCh := make(chan netlink.RouteUpdate, 1)
routeCh <- netlink.RouteUpdate{
Type: syscall.RTM_DELROUTE,
Route: netlink.Route{Dst: defaultRoute, LinkIndex: defaultIF.Index},
}
linkCh := make(chan netlink.LinkUpdate)
err = monitorDefaultInterfaceForTest(t, context.Background(), defaultIF, routeCh, linkCh)
if err == nil || !strings.Contains(err.Error(), "default route deleted") {
t.Fatalf("monitor error = %v, want a default route deletion error", err)
}
})
}
}
func TestMonitorDefaultInterfaceReturnsErrorWhenLinkGoesDown(t *testing.T) {
defaultIF := &net.Interface{Index: 7, Name: "test0"}
routeCh := make(chan netlink.RouteUpdate)
linkCh := make(chan netlink.LinkUpdate, 1)
linkCh <- netlink.LinkUpdate{
Link: &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Index: defaultIF.Index}},
}
err := monitorDefaultInterfaceForTest(t, context.Background(), defaultIF, routeCh, linkCh)
if err == nil {
t.Fatal("expected an error when the default interface goes down")
}
if !strings.Contains(err.Error(), "default interface \"test0\" is down") {
t.Fatalf("unexpected error: %v", err)
}
}
func TestMonitorDefaultInterfaceHandlesClosedSubscriptions(t *testing.T) {
for _, test := range []struct {
name string
closeRoute bool
cancel bool
wantErr string
}{
{name: "closed route subscription", closeRoute: true, wantErr: "route subscription closed"},
{name: "closed link subscription", wantErr: "link subscription closed"},
{name: "context cancellation with closed route subscription", closeRoute: true, cancel: true},
{name: "context cancellation with closed link subscription", cancel: true},
} {
t.Run(test.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
routeCh := make(chan netlink.RouteUpdate)
linkCh := make(chan netlink.LinkUpdate)
if test.closeRoute {
close(routeCh)
} else {
close(linkCh)
}
if test.cancel {
cancel()
}
err := monitorDefaultInterfaceForTest(t, ctx, &net.Interface{}, routeCh, linkCh)
if test.wantErr == "" {
if err != nil {
t.Fatalf("monitor error = %v, want nil", err)
}
return
}
if err == nil || !strings.Contains(err.Error(), test.wantErr) {
t.Fatalf("monitor error = %v, want %q", err, test.wantErr)
}
})
}
}
func monitorDefaultInterfaceForTest(t *testing.T, ctx context.Context, defaultIF *net.Interface,
routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) error {
t.Helper()
errCh := make(chan error, 1)
go func() {
errCh <- monitorDefaultInterface(ctx, defaultIF, routeCh, linkCh)
}()
select {
case err := <-errCh:
return err
case <-time.After(time.Second):
t.Fatal("default interface monitor did not return")
return nil
}
}
func TestMonitorDefaultInterfaceHandlesClosedSubscriptionsAfterCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
routeCh := make(chan netlink.RouteUpdate)
linkCh := make(chan netlink.LinkUpdate)
errCh := make(chan error, 1)
go func() {
errCh <- monitorDefaultInterface(ctx, &net.Interface{}, routeCh, linkCh)
}()
cancel()
close(routeCh)
close(linkCh)
select {
case err := <-errCh:
if err != nil {
t.Fatalf("monitor error = %v, want nil", err)
}
case <-time.After(time.Second):
t.Fatal("monitor did not stop after cancellation")
}
}
func TestMonitorDefaultInterfaceRetriesClosedSubscription(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
secondSubscribed := make(chan struct{})
attempts := 0
subscribe := func(ctx context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error) {
if err := ctx.Err(); err != nil {
return nil, nil, err
}
attempts++
routeCh := make(chan netlink.RouteUpdate)
linkCh := make(chan netlink.LinkUpdate)
if attempts == 1 {
close(routeCh)
close(linkCh)
return routeCh, linkCh, nil
}
close(secondSubscribed)
go func() {
<-ctx.Done()
close(routeCh)
close(linkCh)
}()
return routeCh, linkCh, nil
}
lookup := func() (*net.Interface, error) {
return &net.Interface{Index: 2, Name: "refreshed"}, nil
}
done := make(chan error, 1)
go func() {
done <- monitorDefaultInterfaceWithRetry(ctx, &net.Interface{Index: 1, Name: "original"}, subscribe, lookup, time.Millisecond)
}()
select {
case <-secondSubscribed:
case <-time.After(time.Second):
t.Fatal("monitor did not resubscribe after channel closure")
}
cancel()
select {
case err := <-done:
if err != nil {
t.Fatalf("monitor returned an error after cancellation: %v", err)
}
case <-time.After(time.Second):
t.Fatal("resubscribed monitor did not stop after cancellation")
}
if attempts != 2 {
t.Fatalf("subscription attempts = %d, want 2", attempts)
}
}
func TestMonitorDefaultInterfaceIgnoresNilLink(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
routeCh := make(chan netlink.RouteUpdate)
linkCh := make(chan netlink.LinkUpdate)
errCh := make(chan error, 1)
go func() {
errCh <- monitorDefaultInterface(ctx, &net.Interface{}, routeCh, linkCh)
}()
linkCh <- netlink.LinkUpdate{}
cancel()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("monitor error = %v, want nil", err)
}
case <-time.After(time.Second):
t.Fatal("monitor did not stop after nil link update")
}
}
func TestDrainDefaultInterfaceSubscriptionsIsBounded(t *testing.T) {
done := make(chan struct{})
go func() {
drainDefaultInterfaceSubscriptions(make(chan netlink.RouteUpdate), make(chan netlink.LinkUpdate))
close(done)
}()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("subscription drain did not stop")
}
}
func TestMonitorDefaultInterfaceReturnsErrorWhenTestLinkIsSetDown(t *testing.T) {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
originalNS, err := netns.Get()
if err != nil {
t.Fatalf("getting current network namespace: %v", err)
}
defer originalNS.Close()
testNS, err := netns.New()
if err != nil {
if requireNetworkNamespaces {
t.Fatalf("creating isolated network namespace: %v", err)
}
t.Skipf("creating isolated network namespace: %v", err)
}
defer testNS.Close()
defer func() {
if err := netns.Set(originalNS); err != nil {
t.Errorf("restoring network namespace: %v", err)
}
}()
link := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kv-monitor0"}}
if err := netlink.LinkAdd(link); err != nil {
t.Fatalf("creating test interface: %v", err)
}
defer func() {
if err := netlink.LinkDel(link); err != nil {
t.Errorf("deleting test interface: %v", err)
}
}()
if err := netlink.LinkSetUp(link); err != nil {
t.Fatalf("bringing test interface up: %v", err)
}
defaultIF, err := net.InterfaceByName(link.Attrs().Name)
if err != nil {
t.Fatalf("getting test interface: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
routeCh, linkCh, err := subscribeDefaultInterface(ctx)
if err != nil {
t.Fatalf("subscribing to link updates: %v", err)
}
errCh := make(chan error, 1)
go func() {
errCh <- monitorDefaultInterface(ctx, defaultIF, routeCh, linkCh)
}()
if err := netlink.LinkSetDown(link); err != nil {
t.Fatalf("bringing test interface down: %v", err)
}
select {
case err := <-errCh:
if err == nil || !strings.Contains(err.Error(), "default interface \"kv-monitor0\" is down") {
t.Fatalf("monitor error = %v, want default-interface-down error", err)
}
case <-time.After(time.Second):
t.Fatal("monitor did not report the interface going down")
}
}