From be536eaaf73c80fa5161e757ac18b472498f986e Mon Sep 17 00:00:00 2001 From: Marcel Fest Date: Fri, 18 Sep 2026 14:50:37 +0200 Subject: [PATCH] fix: wireguard + routing-protocol sync (#1769) * fix(kubevip): reject out-of-range routing protocol values Netlink carries the address and route protocol in a single byte, so a configured value above 255 was silently truncated on the wire and never matched again on readback. Reject it during config validation instead. Signed-off-by: Marcel Fest * feat(wireguard): qualify service tunnel IDs by protocol Sanitisation maps '-' onto the '_' separator, so "a-b/c" and "a/b-c" shared one nftables chain, and TCP and UDP on the same port collided. ServicePortIDs appends the protocol and, when sanitisation changed the name or the ID grew too long, a hash of the raw name. It also returns the previous port-only ID so existing chains can be migrated. Signed-off-by: Marcel Fest * fix(arp): guard manager state behind a single mutex Instances were kept in a sync.Map with a per-instance mutex for the refcount, so lookup and refcount update were not atomic: concurrent Insert and Remove could resurrect a deleted instance or drop a live one. Hold one manager mutex across both, and buffer link subscriptions so a netlink sender is never parked on an unread channel during shutdown. Signed-off-by: Marcel Fest * fix(instance): roll back partially created instances Instance creation added addresses, VLAN or macvlan links and DHCP clients incrementally, so a failure part way through left the node holding state nobody owned. Initialization now unwinds what it created, and link cleanup only deletes attachments this instance created that no remaining instance still uses. Namespace-dependent tests now skip unless KUBE_VIP_REQUIRE_NETNS is set, which CI sets on the privileged job so lost capabilities turn it red instead of silently skipping. Signed-off-by: Marcel Fest * test(e2e): give docker kill more time under parallel load The ARP suite runs four kind clusters against one Docker daemon, so acknowledging a leader kill regularly exceeded the 5s budget and failed the IPv6 failover specs before any assertion ran. Signed-off-by: Marcel Fest * fix: regression on preserveOnLeadershipLoss Signed-off-by: Marcel Fest * fix: use the introduced wireguard service_id Signed-off-by: Marcel Fest * fix(services): reuse link attachment ownership on service delete deleteService removed VLAN and macvlan links unconditionally, which tore down interfaces kube-vip had only adopted and interfaces another Service still used. Route the delete path through CleanupLinkAttachments and pass the remaining instances so ownership is handed over instead. Signed-off-by: Marcel Fest --------- Signed-off-by: Marcel Fest --- .github/workflows/ci.yaml | 4 + pkg/arp/arp.go | 114 +++++---- pkg/arp/arp_test.go | 119 +++++++++ pkg/endpoints/endpoints_test.go | 4 +- pkg/endpoints/endpoints_wireguard.go | 39 ++- pkg/instance/instance.go | 355 ++++++++++++++++++-------- pkg/instance/instance_test.go | 65 +++++ pkg/instance/links_linux_test.go | 102 ++++++++ pkg/kubevip/config_validation.go | 14 + pkg/kubevip/config_validation_test.go | 24 ++ pkg/services/processor_lease_test.go | 4 +- pkg/services/services.go | 33 +-- pkg/services/services_wireguard.go | 74 +++--- pkg/wireguard/service_id.go | 50 ++++ pkg/wireguard/service_id_test.go | 44 ++++ testing/e2e/e2e_test.go | 3 +- 16 files changed, 789 insertions(+), 259 deletions(-) create mode 100644 pkg/arp/arp_test.go create mode 100644 pkg/instance/instance_test.go create mode 100644 pkg/instance/links_linux_test.go create mode 100644 pkg/wireguard/service_id.go create mode 100644 pkg/wireguard/service_id_test.go diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 97ea085b..d84c66d2 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -51,6 +51,10 @@ jobs: go-version-file: go.mod - name: Run tests run: make unit-tests + - name: Run privileged network tests + run: | + sudo -E env PATH="$PATH" KUBE_VIP_REQUIRE_NETNS=1 go test -race ./pkg/services ./pkg/vip ./pkg/instance \ + -run 'TestRecover|TestServiceAddressRetained|TestRetainControlPlaneVIPs|TestAddressProtocol|TestKubeVIPAddressProtocol|TestCleanupKubeVIPAddresses|TestMonitorDefaultInterfaceReturnsErrorWhenTestLinkIsSetDown|TestCleanupLinkAttachmentsOnlyDeletesOwnedVLAN' integration-tests: name: Integration tests runs-on: ubuntu-latest diff --git a/pkg/arp/arp.go b/pkg/arp/arp.go index 53c457ea..ba9c38cc 100644 --- a/pkg/arp/arp.go +++ b/pkg/arp/arp.go @@ -13,15 +13,17 @@ import ( "github.com/vishvananda/netlink" ) +const linkSubscriptionBuffer = 64 + type Manager struct { - instances sync.Map + mu sync.Mutex + instances map[string]*Instance config *kubevip.Config } type Instance struct { network vip.Network ndp *vip.NdpResponder - mu sync.Mutex counter int } @@ -31,7 +33,8 @@ func NewManager(config *kubevip.Config) *Manager { config.ArpBroadcastRate = 3000 } return &Manager{ - config: config, + instances: make(map[string]*Instance), + config: config, } } @@ -48,19 +51,16 @@ func (i *Instance) Name() string { } func (m *Manager) Insert(instance *Instance) { - i, err := m.get(instance.Name()) - if err != nil { - log.Error("[ARP manager] unable to insert instance", "err", err) + m.mu.Lock() + defer m.mu.Unlock() + + existing := m.instances[instance.Name()] + if existing == nil { + m.instances[instance.Name()] = instance + log.Info("[ARP manager] inserting ARP/NDP instance", "name", instance.Name()) return } - if i == nil { - log.Info("[ARP manager] inserting ARP/NDP instance", "name", instance.Name()) - m.instances.Store(instance.Name(), instance) - } else { - i.mu.Lock() - defer i.mu.Unlock() - i.counter++ - } + existing.counter++ } func (m *Manager) Remove(instance *Instance) { @@ -77,14 +77,11 @@ func (m *Manager) RemoveOnLeadershipLoss(instance *Instance) { } func (m *Manager) RemoveWithIPDelete(instance *Instance, deleteIP bool) { - i, err := m.get(instance.Name()) - if err != nil { - log.Error("[ARP manager] unable to remove the instance", "err", err) - return - } + m.mu.Lock() + defer m.mu.Unlock() + + i := m.instances[instance.Name()] if i != nil { - i.mu.Lock() - defer i.mu.Unlock() i.counter-- if i.counter == 0 { log.Info("[ARP manager] removing ARP/NDP instance", "name", instance.Name()) @@ -93,7 +90,7 @@ func (m *Manager) RemoveWithIPDelete(instance *Instance, deleteIP bool) { log.Error("failed to delete IP", "address", instance.network.IP(), "err", err) } } - m.instances.Delete(instance.Name()) + delete(m.instances, instance.Name()) } } else { log.Warn("[ARP manager] unable to remove the instance - instance not found", "name", instance.Name()) @@ -101,14 +98,11 @@ func (m *Manager) RemoveWithIPDelete(instance *Instance, deleteIP bool) { } func (m *Manager) Count(name string) int { - i, err := m.get(name) - if err != nil { - log.Error("[ARP manager] unable to count instance", "err", err) - return -1 - } + m.mu.Lock() + defer m.mu.Unlock() + + i := m.instances[name] if i != nil { - i.mu.Lock() - defer i.mu.Unlock() return i.counter } return 0 @@ -158,35 +152,22 @@ func (m *Manager) StartAdvertisement(ctx context.Context, killFunc func()) { case <-ctx.Done(): // if cancel() execute return case <-ticker.C: // send gratuitous ARP/NDP on each tick - m.instances.Range(func(_ any, instance any) bool { - if i, ok := instance.(*Instance); ok { - i.mu.Lock() - defer i.mu.Unlock() - if i.counter > 0 { - ensureIPAndSendGratuitous(i) - } else { - // this instance should not be advertised - delete the IP just in case... - if _, err := i.network.DeleteIP(); err != nil { - log.Error("[ARP manager] failed to delete IP", "address", i.network.IP(), "err", err) - } - } - } - return true - }) + m.advertiseAll() } } } -func (m *Manager) get(name string) (*Instance, error) { - i, exists := m.instances.Load(name) - if !exists { - return nil, nil +func (m *Manager) advertiseAll() { + m.mu.Lock() + defer m.mu.Unlock() + + for _, instance := range m.instances { + if instance.counter > 0 { + ensureIPAndSendGratuitous(instance) + } else if _, err := instance.network.DeleteIP(); err != nil { + log.Error("[ARP manager] failed to delete IP", "address", instance.network.IP(), "err", err) + } } - inst, ok := i.(*Instance) - if !ok { - return nil, fmt.Errorf("value for name %q is not of Instance pointer type", name) - } - return inst, nil } // ensureIPAndSendGratuitous - adds IP to the interface if missing, and send @@ -255,13 +236,18 @@ func watch(ctx context.Context, interfaceName string, operStateHandler func(netl return fmt.Errorf("interface %s is not physical, ignoring", interfaceName) } - events := make(chan netlink.LinkUpdate) + // The subscription is buffered and drained on exit: netlink parks its reader + // goroutine on an unread send, which closing done alone does not release. + events := make(chan netlink.LinkUpdate, linkSubscriptionBuffer) done := make(chan struct{}) if err := netlink.LinkSubscribe(events, done); err != nil { return fmt.Errorf("failed to subscribe to the interface events: %w", err) } - defer close(done) + defer func() { + close(done) + drainLinkUpdates(events) + }() // handle initial state operStateHandler(ifname.Attrs().OperState) @@ -290,3 +276,21 @@ func watch(ctx context.Context, interfaceName string, operStateHandler func(netl func isUp(operState netlink.LinkOperState) bool { return operState == netlink.OperUp } + +// drainLinkUpdates releases a netlink sender that is parked on an unread update +// so its goroutine can observe the closed subscription and exit. +func drainLinkUpdates(events <-chan netlink.LinkUpdate) { + timer := time.NewTimer(100 * time.Millisecond) + defer timer.Stop() + + for { + select { + case _, ok := <-events: + if !ok { + return + } + case <-timer.C: + return + } + } +} diff --git a/pkg/arp/arp_test.go b/pkg/arp/arp_test.go new file mode 100644 index 00000000..3e487498 --- /dev/null +++ b/pkg/arp/arp_test.go @@ -0,0 +1,119 @@ +package arp + +import ( + "sync" + "testing" + "time" + + "github.com/kube-vip/kube-vip/pkg/kubevip" + "github.com/vishvananda/netlink" + v1 "k8s.io/api/core/v1" +) + +func TestDrainLinkUpdatesReleasesParkedSender(t *testing.T) { + events := make(chan netlink.LinkUpdate) + sent := make(chan struct{}) + go func() { + events <- netlink.LinkUpdate{} + close(sent) + }() + + drainLinkUpdates(events) + + select { + case <-sent: + case <-time.After(time.Second): + t.Fatal("netlink sender is still parked on an unread link update") + } +} + +// stubNetwork is a minimal vip.Network implementation; only ARPName matters here. +type stubNetwork struct { + name string + deleteStarted chan struct{} + releaseDelete chan struct{} +} + +func (s *stubNetwork) AddIP(bool, bool, ...int) (bool, error) { return false, nil } +func (s *stubNetwork) AddRoute(bool) (bool, error) { return false, nil } +func (s *stubNetwork) ReplaceRoute() error { return nil } +func (s *stubNetwork) DeleteIP() (bool, error) { + if s.deleteStarted != nil { + close(s.deleteStarted) + <-s.releaseDelete + } + return true, nil +} +func (s *stubNetwork) DeleteRoute() error { return nil } +func (s *stubNetwork) UpdateRoutes() (bool, error) { return false, nil } +func (s *stubNetwork) IsSet() (*netlink.Addr, error) { return nil, nil } +func (s *stubNetwork) IP() string { return "" } +func (s *stubNetwork) CIDR() string { return "" } +func (s *stubNetwork) IPisLinkLocal() bool { return false } +func (s *stubNetwork) PrepareRoute() *netlink.Route { return nil } +func (s *stubNetwork) RouteHash() string { return "" } +func (s *stubNetwork) SetIP(string) error { return nil } +func (s *stubNetwork) SetServicePorts(*v1.Service) {} +func (s *stubNetwork) Interface() string { return "eth0" } +func (s *stubNetwork) IsDADFAIL() bool { return false } +func (s *stubNetwork) IsDNS() bool { return false } +func (s *stubNetwork) IsDDNS() bool { return false } +func (s *stubNetwork) DDNSHostName() string { return "" } +func (s *stubNetwork) DNSName() string { return "" } +func (s *stubNetwork) SetMask(string) error { return nil } +func (s *stubNetwork) SetHasEndpoints(bool) {} +func (s *stubNetwork) HasEndpoints() bool { return false } +func (s *stubNetwork) ARPName() string { return s.name } +func (s *stubNetwork) GetPossibleSubnets() string { return "" } +func (s *stubNetwork) DHCPFamily() string { return "" } +func (s *stubNetwork) IPVSMark() uint32 { return 0 } + +// TestManagerInsertConcurrentFirstRegistrationsDoNotLoseClaims guards the +// get-then-store race: two never-before-seen instances for the same ARP name +// registering concurrently must both be counted, not just the last writer. +func TestManagerInsertConcurrentFirstRegistrationsDoNotLoseClaims(t *testing.T) { + m := NewManager(&kubevip.Config{ArpBroadcastRate: 3000}) + const concurrent = 8 + + var wg sync.WaitGroup + for range concurrent { + wg.Add(1) + go func() { + defer wg.Done() + m.Insert(NewInstance(&stubNetwork{name: "shared"}, nil)) + }() + } + wg.Wait() + + if got := m.Count("shared"); got != concurrent { + t.Fatalf("Count() = %d, want %d claims registered", got, concurrent) + } +} + +func TestManagerInsertDoesNotJoinEntryBeingRemoved(t *testing.T) { + m := NewManager(&kubevip.Config{ArpBroadcastRate: 3000}) + deleteStarted := make(chan struct{}) + releaseDelete := make(chan struct{}) + first := NewInstance(&stubNetwork{name: "shared", deleteStarted: deleteStarted, releaseDelete: releaseDelete}, nil) + m.Insert(first) + + removeDone := make(chan struct{}) + go func() { + m.Remove(first) + close(removeDone) + }() + <-deleteStarted + + insertDone := make(chan struct{}) + go func() { + m.Insert(NewInstance(&stubNetwork{name: "shared"}, nil)) + close(insertDone) + }() + close(releaseDelete) + <-removeDone + <-insertDone + + if got := m.Count("shared"); got != 1 { + t.Fatalf("Count() = %d, want replacement claim registered", got) + } +} diff --git a/pkg/endpoints/endpoints_test.go b/pkg/endpoints/endpoints_test.go index 6825d36a..ab6f331f 100644 --- a/pkg/endpoints/endpoints_test.go +++ b/pkg/endpoints/endpoints_test.go @@ -102,7 +102,7 @@ func TestUpdateAnnotationsZeroEndpointsThenSameEndpoint(t *testing.T) { service := &v1.Service{ObjectMeta: metav1.ObjectMeta{ Name: "test-service", Namespace: "default", UID: "test-uid", Annotations: annotations, }} - serviceInstance := &instance.Instance{ServiceSnapshot: service.DeepCopy()} + serviceInstance := &instance.Instance{ServiceUID: service.UID, ServiceSnapshot: service.DeepCopy()} instances := []*instance.Instance{serviceInstance} recorder := &recordingProvider{Provider: provider} processor := &Processor{ @@ -167,7 +167,7 @@ func TestUpdateAnnotationsEndpointSlicesClearsConfiguredFamily(t *testing.T) { service := &v1.Service{ObjectMeta: metav1.ObjectMeta{ Name: "test-service", Namespace: "default", UID: "test-uid", Annotations: annotations, }} - instances := []*instance.Instance{{ServiceSnapshot: service.DeepCopy()}} + instances := []*instance.Instance{{ServiceUID: service.UID, ServiceSnapshot: service.DeepCopy()}} recorder := &recordingProvider{Provider: providers.NewEndpointslices()} processor := &Processor{ config: &kubevip.Config{EnableEndpoints: false}, diff --git a/pkg/endpoints/endpoints_wireguard.go b/pkg/endpoints/endpoints_wireguard.go index c5d3eed6..593184b0 100644 --- a/pkg/endpoints/endpoints_wireguard.go +++ b/pkg/endpoints/endpoints_wireguard.go @@ -78,9 +78,6 @@ func (w *wireguardWorker) processInstance(svcCtx *servicecontext.Context, servic return fmt.Errorf("failed to get service IPs: %w", err) } - // Create service identifier - serviceID := utils.SanitizeServiceID(fmt.Sprintf("%s_%s", service.Namespace, service.Name)) - log.Info("[wireguard] updating DNAT rules for endpoint change", "service", service.Name, "namespace", service.Namespace, @@ -123,7 +120,7 @@ func (w *wireguardWorker) processInstance(svcCtx *servicecontext.Context, servic } wgInterface := tunnelConfig.InterfaceName - portServiceID := fmt.Sprintf("%s_p%d", serviceID, port.Port) + portServiceID, _ := wireguard.ServicePortIDs(service.Namespace, service.Name, port) log.Info("[wireguard] applying DNAT rule with load balancing", "service", service.Name, @@ -170,8 +167,6 @@ func (w *wireguardWorker) processInstance(svcCtx *servicecontext.Context, servic func (w *wireguardWorker) clear(svcCtx *servicecontext.Context, lastKnownGoodEndpoint *string, service *v1.Service) { log.Info("[wireguard] clearing DNAT rules (no endpoints)", "service", service.Name, "namespace", service.Namespace) - serviceID := utils.SanitizeServiceID(fmt.Sprintf("%s_%s", service.Namespace, service.Name)) - // Get service IPs to determine IPv4 vs IPv6 serviceIPs, _ := utils.FetchServiceIPs(service) @@ -181,8 +176,6 @@ func (w *wireguardWorker) clear(svcCtx *servicecontext.Context, lastKnownGoodEnd continue } - portServiceID := fmt.Sprintf("%s_p%d", serviceID, port.Port) - // Determine if we have IPv4 or IPv6 hasIPv4, hasIPv6 := false, false for _, vip := range serviceIPs { @@ -193,21 +186,25 @@ func (w *wireguardWorker) clear(svcCtx *servicecontext.Context, lastKnownGoodEnd } } - if hasIPv4 { - if err := nftables.DeleteIngressChains(false, portServiceID); err != nil { - log.Warn("[wireguard] failed to delete IPv4 DNAT chains", - "service", service.Name, - "port", port.Port, - "err", err) + for _, portServiceID := range wireguard.ServicePortIDSet(service.Namespace, service.Name, port) { + if hasIPv4 { + if err := nftables.DeleteIngressChains(false, portServiceID); err != nil { + log.Warn("[wireguard] failed to delete IPv4 DNAT chains", + "service", service.Name, + "port", port.Port, + "id", portServiceID, + "err", err) + } } - } - if hasIPv6 { - if err := nftables.DeleteIngressChains(true, portServiceID); err != nil { - log.Warn("[wireguard] failed to delete IPv6 DNAT chains", - "service", service.Name, - "port", port.Port, - "err", err) + if hasIPv6 { + if err := nftables.DeleteIngressChains(true, portServiceID); err != nil { + log.Warn("[wireguard] failed to delete IPv6 DNAT chains", + "service", service.Name, + "port", port.Port, + "id", portServiceID, + "err", err) + } } } } diff --git a/pkg/instance/instance.go b/pkg/instance/instance.go index dd9c0b11..1afbf7d5 100644 --- a/pkg/instance/instance.go +++ b/pkg/instance/instance.go @@ -2,17 +2,20 @@ package instance import ( "context" + "errors" "fmt" "net" "slices" "strconv" "strings" "sync" + "sync/atomic" log "log/slog" "github.com/vishvananda/netlink" v1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" "github.com/kube-vip/kube-vip/pkg/arp" "github.com/kube-vip/kube-vip/pkg/cluster" @@ -46,18 +49,21 @@ type Instance struct { DHCPv6Client vip.DHCPClient macvlanName string dhcpBroadcast bool + dhcpInterfaceOwned atomic.Bool // Service use Vlan IsVLAN bool VLANInterface string + vlanOwned atomic.Bool // External Gateway IP the service is forwarded from UPNPGatewayIPs []string // Kubernetes service mapping - ServiceSnapshot *v1.Service - - dnsAddresses []string + ServiceUID types.UID + ServiceAddresses []string + ServiceSnapshot *v1.Service + cleanupInfo *ServiceCleanupInfo // AddCalled determined that ActionAdd was already performed for the instance AddCalled bool @@ -67,6 +73,48 @@ type Instance struct { LabelAdded bool } +func (instance *Instance) UID() types.UID { + return instance.ServiceUID +} + +func (instance *Instance) Addresses() []string { + if instance.ServiceAddresses != nil { + return append([]string(nil), instance.ServiceAddresses...) + } + addresses, _ := FetchServiceAddresses(instance.ServiceSnapshot) + return addresses +} + +type ServiceCleanupInfo struct { + Namespace string + Name string + Lease string + ExternalTrafficPolicy v1.ServiceExternalTrafficPolicy +} + +// CleanupInfo returns Service policy captured when the instance was created. +func (instance *Instance) CleanupInfo() (ServiceCleanupInfo, bool) { + if instance == nil { + return ServiceCleanupInfo{}, false + } + if instance.cleanupInfo != nil { + return *instance.cleanupInfo, true + } + if instance.ServiceSnapshot == nil { + return ServiceCleanupInfo{}, false + } + return serviceCleanupInfo(instance.ServiceSnapshot), true +} + +func serviceCleanupInfo(service *v1.Service) ServiceCleanupInfo { + return ServiceCleanupInfo{ + Namespace: service.Namespace, + Name: service.Name, + Lease: service.Annotations[kubevip.ServiceLease], + ExternalTrafficPolicy: service.Spec.ExternalTrafficPolicy, + } +} + type Port struct { Port uint16 Type string @@ -78,16 +126,25 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, instanceAddresses, instanceHostnames := FetchServiceAddresses(svc) log.Info("new instance", "namespace", svc.Namespace, "service", svc.Name, "addresses", instanceAddresses, "hostnames", instanceHostnames) + cleanupInfo := serviceCleanupInfo(svc) + instance := &Instance{ + ServiceUID: svc.UID, + ServiceAddresses: append([]string(nil), instanceAddresses...), + ServiceSnapshot: svc, + cleanupInfo: &cleanupInfo, + } + if err := instance.initialize(ctx, svc, config, intfMgr, arpMgr, routeMgr, nodeLabelMgr, wg, instanceAddresses, instanceHostnames); err != nil { + return nil, errors.Join(err, instance.CleanupLinkAttachments()) + } + return instance, nil +} + +func (instance *Instance) initialize(ctx context.Context, svc *v1.Service, config *kubevip.Config, + intfMgr *networkinterface.Manager, arpMgr *arp.Manager, routeMgr *route.Manager, + nodeLabelMgr node.Labeler, wg *sync.WaitGroup, instanceAddresses, instanceHostnames []string) error { var newVips []*kubevip.Config var link netlink.Link var err error - var dnsAddresses []string - - // Create new service - instance := &Instance{ - ServiceSnapshot: svc, - dnsAddresses: dnsAddresses, - } for _, address := range instanceAddresses { // Detect if we're using a specific interface for services @@ -143,10 +200,10 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, if link == nil { if link, err = netlink.LinkByName(svcInterface); err != nil { - return nil, fmt.Errorf("failed to get interface %s: %w", svcInterface, err) + return fmt.Errorf("failed to get interface %s: %w", svcInterface, err) } if link == nil { - return nil, fmt.Errorf("failed to get interface %s", svcInterface) + return fmt.Errorf("failed to get interface %s", svcInterface) } } @@ -163,7 +220,7 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, } if (config.Address != "" || config.VIP != "") && (ipv4AutoSubnet || ipv6AutoSubnet) { - return nil, fmt.Errorf("auto subnet discovery cannot be used if VIP address was provided") + return fmt.Errorf("auto subnet discovery cannot be used if VIP address was provided") } subnet := "" @@ -172,7 +229,7 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, if ipv4AutoSubnet { subnet, err = autoFindSubnet(link, address) if err != nil { - return nil, fmt.Errorf("failed to automatically find subnet for service %s/%s with IP address %s on interface %s: %w", svc.Namespace, svc.Name, address, svcInterface, err) + return fmt.Errorf("failed to automatically find subnet for service %s/%s with IP address %s on interface %s: %w", svc.Namespace, svc.Name, address, svcInterface, err) } } else { if cidrs[0] != "" && cidrs[0] != kubevip.Auto { @@ -185,7 +242,7 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, if ipv6AutoSubnet { subnet, err = autoFindSubnet(link, address) if err != nil { - return nil, fmt.Errorf("failed to automatically find subnet for service %s/%s with IP address %s on interface %s: %w", svc.Namespace, svc.Name, address, svcInterface, err) + return fmt.Errorf("failed to automatically find subnet for service %s/%s with IP address %s on interface %s: %w", svc.Namespace, svc.Name, address, svcInterface, err) } } else { if len(cidrs) > 1 && cidrs[1] != "" && cidrs[1] != kubevip.Auto { @@ -198,25 +255,26 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, // Generate new Virtual IP configuration newVips = append(newVips, &kubevip.Config{ - VIP: address, - Interface: svcInterface, - SingleNode: true, - EnableARP: config.EnableARP, - EnableBGP: config.EnableBGP, - BGPAttachIPToInterface: config.BGPAttachIPToInterface, - VIPSubnet: subnet, - EnableRoutingTable: config.EnableRoutingTable, - RoutingTableID: config.RoutingTableID, - RoutingTableType: config.RoutingTableType, - RoutingProtocol: config.RoutingProtocol, - SkipDAD: config.SkipDAD, - ArpBroadcastRate: config.ArpBroadcastRate, - EnableServiceSecurity: config.EnableServiceSecurity, - DNSMode: config.DNSMode, - DHCPMode: config.DHCPMode, - DHCPBackoffAttempts: config.DHCPBackoffAttempts, - DisableServiceUpdates: config.DisableServiceUpdates, - EnableServicesElection: config.EnableServicesElection, + VIP: address, + Interface: svcInterface, + SingleNode: true, + EnableARP: config.EnableARP, + EnableBGP: config.EnableBGP, + BGPAttachIPToInterface: config.BGPAttachIPToInterface, + VIPSubnet: subnet, + EnableRoutingTable: config.EnableRoutingTable, + RoutingTableID: config.RoutingTableID, + RoutingTableType: config.RoutingTableType, + RoutingProtocol: config.RoutingProtocol, + SkipDAD: config.SkipDAD, + ArpBroadcastRate: config.ArpBroadcastRate, + EnableServiceSecurity: config.EnableServiceSecurity, + DNSMode: config.DNSMode, + DHCPMode: config.DHCPMode, + DHCPBackoffAttempts: config.DHCPBackoffAttempts, + DisableServiceUpdates: config.DisableServiceUpdates, + EnableServicesElection: config.EnableServicesElection, + // cleanupVIPs reads this from the per-VIP config, so Service VIPs need it too. PreserveVIPOnLeadershipLoss: config.PreserveVIPOnLeadershipLoss, KubernetesLeaderElection: kubevip.KubernetesLeaderElection{ EnableLeaderElection: config.EnableLeaderElection, @@ -256,10 +314,10 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, if link == nil { if link, err = netlink.LinkByName(svcInterface); err != nil { - return nil, fmt.Errorf("failed to get interface %s: %w", svcInterface, err) + return fmt.Errorf("failed to get interface %s: %w", svcInterface, err) } if link == nil { - return nil, fmt.Errorf("failed to get interface %s", svcInterface) + return fmt.Errorf("failed to get interface %s", svcInterface) } } @@ -295,7 +353,7 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, if requestedIP != "" { requestedIPs := strings.Split(requestedIP, ",") if len(requestedIPs) > 2 { - return nil, fmt.Errorf("annotation %q cannot request more than one IPv4 and one Ipv6 address", kubevip.RequestedIP) + return fmt.Errorf("annotation %q cannot request more than one IPv4 and one Ipv6 address", kubevip.RequestedIP) } for _, ip := range requestedIPs { netip := net.ParseIP(ip) @@ -334,42 +392,40 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, // If this was purposely created with the address '0.0.0.0', or '::' // we will create a macvlan on the main interface and a DHCP client if len(instanceAddresses) > 2 && (slices.Contains(instanceAddresses, "0.0.0.0") || slices.Contains(instanceAddresses, "::")) { - return nil, fmt.Errorf("DHCP cannot be used if more than 2 addresses (one IPv4 and one IPv6) were specified") + return fmt.Errorf("DHCP cannot be used if more than 2 addresses (one IPv4 and one IPv6) were specified") } - for i := range instance.VIPConfigs { - if instance.VIPConfigs[i].VIP == "0.0.0.0" { - err := instance.startDHCP(ctx, i, config.DHCPBackoffAttempts, wg) + for index := range instance.VIPConfigs { + if instance.VIPConfigs[index].VIP == "0.0.0.0" { + err := instance.startDHCP(ctx, index, config.DHCPBackoffAttempts, wg) if err != nil { - return nil, err + return err } select { case <-ctx.Done(): - return nil, fmt.Errorf("context error while starting DHCPv4 for %s/%s: error: %w", - instance.ServiceSnapshot.Namespace, instance.ServiceSnapshot.Name, ctx.Err()) + return ctx.Err() case err := <-instance.DHCPv4Client.ErrorChannel(): - return nil, fmt.Errorf("error starting DHCPv4 for %s/%s: error: %s", + return fmt.Errorf("error starting DHCPv4 for %s/%s: error: %s", instance.ServiceSnapshot.Namespace, instance.ServiceSnapshot.Name, err) case ip := <-instance.DHCPv4Client.IPChannel(): - instance.VIPConfigs[i].Interface = instance.DHCPInterface - instance.VIPConfigs[i].VIP = ip + instance.VIPConfigs[index].Interface = instance.DHCPInterface + instance.VIPConfigs[index].VIP = ip instance.DHCPInterfaceIPv4 = ip } } - if instance.VIPConfigs[i].VIP == "::" { - err := instance.startDHCP(ctx, i, config.DHCPBackoffAttempts, wg) + if instance.VIPConfigs[index].VIP == "::" { + err := instance.startDHCP(ctx, index, config.DHCPBackoffAttempts, wg) if err != nil { - return nil, err + return err } select { case <-ctx.Done(): - return nil, fmt.Errorf("context error while starting DHCPv6 for %s/%s: error: %w", - instance.ServiceSnapshot.Namespace, instance.ServiceSnapshot.Name, ctx.Err()) + return ctx.Err() case err := <-instance.DHCPv6Client.ErrorChannel(): - return nil, fmt.Errorf("error starting DHCPv6 for %s/%s: error: %s", + return fmt.Errorf("error starting DHCPv6 for %s/%s: error: %s", instance.ServiceSnapshot.Namespace, instance.ServiceSnapshot.Name, err) case ip := <-instance.DHCPv6Client.IPChannel(): - instance.VIPConfigs[i].Interface = instance.DHCPInterface - instance.VIPConfigs[i].VIP = ip + instance.VIPConfigs[index].Interface = instance.DHCPInterface + instance.VIPConfigs[index].VIP = ip instance.DHCPInterfaceIPv6 = ip } } @@ -377,56 +433,56 @@ func NewInstance(ctx context.Context, svc *v1.Service, config *kubevip.Config, ddnsAnnotation, exists := svc.Annotations[kubevip.ServiceDDNS] if exists { - instance.VIPConfigs[i].DDNS, err = strconv.ParseBool(ddnsAnnotation) + instance.VIPConfigs[index].DDNS, err = strconv.ParseBool(ddnsAnnotation) if err != nil { log.Error("Failed to add service", "err", err) - return nil, err + return err } } if len(svc.Spec.IPFamilies) > 0 { if len(svc.Spec.IPFamilies) > 1 { - instance.VIPConfigs[i].DHCPMode = utils.DualFamily - instance.VIPConfigs[i].DNSMode = utils.DualFamily + instance.VIPConfigs[index].DHCPMode = utils.DualFamily + instance.VIPConfigs[index].DNSMode = utils.DualFamily switch *svc.Spec.IPFamilyPolicy { case v1.IPFamilyPolicyRequireDualStack: - instance.VIPConfigs[i].IsDualStack = true - instance.VIPConfigs[i].RequireDualStack = true + instance.VIPConfigs[index].IsDualStack = true + instance.VIPConfigs[index].RequireDualStack = true case v1.IPFamilyPolicyPreferDualStack: - instance.VIPConfigs[i].IsDualStack = true - instance.VIPConfigs[i].RequireDualStack = false + instance.VIPConfigs[index].IsDualStack = true + instance.VIPConfigs[index].RequireDualStack = false default: - instance.VIPConfigs[i].IsDualStack = false - instance.VIPConfigs[i].RequireDualStack = false + instance.VIPConfigs[index].IsDualStack = false + instance.VIPConfigs[index].RequireDualStack = false } } else { if strings.EqualFold(string(svc.Spec.IPFamilies[0]), utils.IPv4Family) { - instance.VIPConfigs[i].DHCPMode = strings.ToLower(utils.IPv4Family) - instance.VIPConfigs[i].DNSMode = strings.ToLower(utils.IPv4Family) + instance.VIPConfigs[index].DHCPMode = strings.ToLower(utils.IPv4Family) + instance.VIPConfigs[index].DNSMode = strings.ToLower(utils.IPv4Family) } else { - instance.VIPConfigs[i].DHCPMode = strings.ToLower(utils.IPv6Family) - instance.VIPConfigs[i].DNSMode = strings.ToLower(utils.IPv6Family) + instance.VIPConfigs[index].DHCPMode = strings.ToLower(utils.IPv6Family) + instance.VIPConfigs[index].DNSMode = strings.ToLower(utils.IPv6Family) } } } - instance.VIPConfigs[i].EgressWithNftables = config.EgressWithNftables + instance.VIPConfigs[index].EgressWithNftables = config.EgressWithNftables - c, err := cluster.InitCluster(instance.VIPConfigs[i], false, intfMgr, arpMgr, routeMgr, nodeLabelMgr) + c, err := cluster.InitCluster(instance.VIPConfigs[index], false, intfMgr, arpMgr, routeMgr, nodeLabelMgr) if err != nil { log.Error("failed to add service", "err", err) - return nil, err + return err } - for i := range c.Network { - c.Network[i].SetServicePorts(svc) + for networkIndex := range c.Network { + c.Network[networkIndex].SetServicePorts(svc) } instance.Clusters = append(instance.Clusters, c) - log.Info("(svcs) adding VIP", "ip", instance.VIPConfigs[i].VIP, "interface", instance.VIPConfigs[i].Interface, "namespace", svc.Namespace, "name", svc.Name) + log.Info("(svcs) adding VIP", "ip", instance.VIPConfigs[index].VIP, "interface", instance.VIPConfigs[index].Interface, "namespace", svc.Namespace, "name", svc.Name) } - return instance, nil + return nil } func autoFindInterface(ip string) (netlink.Link, error) { @@ -486,16 +542,17 @@ func getAutoInterfaceName(link netlink.Link, defaultInterface string) string { return link.Attrs().Name } -func (i *Instance) addVLAN(parentInterface string, tag int) error { - var parent netlink.Link - +func (instance *Instance) addVLAN(parentInterface string, tag int) error { interfaceName := fmt.Sprintf("%s.%d", parentInterface, tag) + parent, err := netlink.LinkByName(parentInterface) + if err != nil { + return fmt.Errorf("finding VLAN parent interface %s: %w", parentInterface, err) + } iface, err := netlink.LinkByName(interfaceName) if err != nil { - // check if parent interface doesnt exist - parent, err = netlink.LinkByName(parentInterface) - if err != nil { - return fmt.Errorf("error finding VLAN parent interface %s: %v", parentInterface, err) + var notFound netlink.LinkNotFoundError + if !errors.As(err, ¬Found) { + return fmt.Errorf("finding VLAN interface %s: %w", interfaceName, err) } log.Info("Creating new VLAN interface", "interface", interfaceName) @@ -513,6 +570,7 @@ func (i *Instance) addVLAN(parentInterface string, tag int) error { if err != nil { return fmt.Errorf("could not add VLAN %s: %v", interfaceName, err) } + instance.vlanOwned.Store(true) err = netlink.LinkSetUp(vlan) if err != nil { @@ -531,26 +589,96 @@ func (i *Instance) addVLAN(parentInterface string, tag int) error { } } - i.VLANInterface = interfaceName - i.IsVLAN = true + instance.VLANInterface = interfaceName + instance.IsVLAN = true return nil } -func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uint, wg *sync.WaitGroup) error { - if len(i.VIPConfigs) > 2 { - return fmt.Errorf("DHCP can be used with 2 VIP config maximally, got: %v", len(i.VIPConfigs)) +// CleanupLinkAttachments stops this instance's DHCP clients and removes only +// VLAN or macvlan links created by this instance that are not used by a +// remaining Service instance. +func (instance *Instance) CleanupLinkAttachments(remaining ...*Instance) error { + var errs []error + if instance.DHCPv4Client != nil { + instance.DHCPv4Client.Stop() } - parent, err := netlink.LinkByName(i.VIPConfigs[index].Interface) + if instance.DHCPv6Client != nil { + instance.DHCPv6Client.Stop() + } + if instance.dhcpInterfaceOwned.Load() { + if transferLinkAttachmentOwnership(instance.DHCPInterface, remaining, false) { + instance.dhcpInterfaceOwned.Store(false) + } else if err := deleteOwnedLink(instance.DHCPInterface, "DHCP"); err != nil { + errs = append(errs, err) + } else { + instance.dhcpInterfaceOwned.Store(false) + } + } + if instance.vlanOwned.Load() { + if transferLinkAttachmentOwnership(instance.VLANInterface, remaining, true) { + instance.vlanOwned.Store(false) + } else if err := deleteOwnedLink(instance.VLANInterface, "VLAN"); err != nil { + errs = append(errs, err) + } else { + instance.vlanOwned.Store(false) + } + } + return errors.Join(errs...) +} + +func transferLinkAttachmentOwnership(name string, instances []*Instance, vlan bool) bool { + if name == "" { + return false + } + for _, instance := range instances { + if instance == nil { + continue + } + if vlan && instance.IsVLAN && instance.VLANInterface == name { + instance.vlanOwned.Store(true) + return true + } + if !vlan && (instance.IsDHCPv4 || instance.IsDHCPv6) && instance.DHCPInterface == name { + instance.dhcpInterfaceOwned.Store(true) + return true + } + } + return false +} + +func deleteOwnedLink(name, kind string) error { + if name == "" { + return nil + } + link, err := netlink.LinkByName(name) + if err != nil { + var notFound netlink.LinkNotFoundError + if errors.As(err, ¬Found) { + return nil + } + return fmt.Errorf("find %s interface %q: %w", kind, name, err) + } + if err := netlink.LinkDel(link); err != nil { + return fmt.Errorf("delete %s interface %q: %w", kind, name, err) + } + return nil +} + +func (instance *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uint, wg *sync.WaitGroup) error { + if len(instance.VIPConfigs) > 2 { + return fmt.Errorf("DHCP can be used with 2 VIP config maximally, got: %v", len(instance.VIPConfigs)) + } + parent, err := netlink.LinkByName(instance.VIPConfigs[index].Interface) if err != nil { return fmt.Errorf("error finding VIP Interface, for building DHCP Link : %v", err) } - interfaceName := i.macvlanName + interfaceName := instance.macvlanName if interfaceName == "" { // Generate name from UID - interfaceName = fmt.Sprintf("vip-%s", i.ServiceSnapshot.UID[0:8]) + interfaceName = fmt.Sprintf("vip-%s", instance.UID()[0:8]) } // Check if the interface doesn't exist first @@ -558,8 +686,8 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin if err != nil { log.Info("creating new macvlan interface for DHCP", "interface", interfaceName) - hwaddr, err := net.ParseMAC(i.DHCPInterfaceHwaddr) - if i.DHCPInterfaceHwaddr != "" && err != nil { + hwaddr, err := net.ParseMAC(instance.DHCPInterfaceHwaddr) + if instance.DHCPInterfaceHwaddr != "" && err != nil { return err } else if hwaddr == nil { hwaddr, err = net.ParseMAC(vip.GenerateMac()) @@ -582,6 +710,7 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin if err != nil { return fmt.Errorf("could not add %s: %v", interfaceName, err) } + instance.dhcpInterfaceOwned.Store(true) err = netlink.LinkSetUp(mac) if err != nil { @@ -597,7 +726,7 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin } var initRebootFlag bool - ip := net.ParseIP(i.VIPConfigs[index].VIP) + ip := net.ParseIP(instance.VIPConfigs[index].VIP) var client vip.DHCPClient if ip.To4() != nil { @@ -605,14 +734,14 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin rpfilterSetting := "0" // Check if we need to set an override rp_filter value for the interface - if i.ServiceSnapshot.Annotations[kubevip.RPFilter] != "" { + if instance.ServiceSnapshot.Annotations[kubevip.RPFilter] != "" { // Check the rp_filter value - rpFilter, err := strconv.Atoi(i.ServiceSnapshot.Annotations[kubevip.RPFilter]) + rpFilter, err := strconv.Atoi(instance.ServiceSnapshot.Annotations[kubevip.RPFilter]) if err != nil { log.Error("[DHCP] unable to process rp_filter", "value", rpFilter) } else { if rpFilter >= 0 && rpFilter < 3 { // Ensure the value is 0,1,2 - rpfilterSetting = i.ServiceSnapshot.Annotations[kubevip.RPFilter] + rpfilterSetting = instance.ServiceSnapshot.Annotations[kubevip.RPFilter] } else { log.Error("[DHCP] rp_filter value not within range 0-2", "value", rpFilter) } @@ -624,38 +753,38 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin log.Error("[DHCP] unable to write rp_filter", "value", rpfilterSetting, "err", err) } - if i.DHCPInterfaceIPv4 != "" { + if instance.DHCPInterfaceIPv4 != "" { initRebootFlag = true } - client = vip.NewDHCPv4Client(iface, initRebootFlag, i.DHCPInterfaceIPv4, backoffAttempts, i.dhcpBroadcast) + client = vip.NewDHCPv4Client(iface, initRebootFlag, instance.DHCPInterfaceIPv4, backoffAttempts, instance.dhcpBroadcast) // Add the client so that we can call it to stop function - i.DHCPv4Client = client + instance.DHCPv4Client = client // Set that DHCPv4 is enabled - i.IsDHCPv4 = true + instance.IsDHCPv4 = true } else { - if i.DHCPInterfaceIPv6 != "" { + if instance.DHCPInterfaceIPv6 != "" { initRebootFlag = true } - client, err = vip.NewDHCPv6Client(iface, parent, initRebootFlag, i.DHCPInterfaceIPv6, backoffAttempts) + client, err = vip.NewDHCPv6Client(iface, parent, initRebootFlag, instance.DHCPInterfaceIPv6, backoffAttempts) if err != nil { return fmt.Errorf("unable to create client: %w", err) } // Add the client so that we can call it to stop function - i.DHCPv6Client = client + instance.DHCPv6Client = client // Set that DHCPv6 is enabled - i.IsDHCPv6 = true + instance.IsDHCPv6 = true } // Add hostname to dhcp client if annotated - if i.DHCPHostname != "" { - log.Info("Hostname specified for dhcp lease", "interface", interfaceName, "hostname", i.DHCPHostname) - client.WithHostName(i.DHCPHostname) + if instance.DHCPHostname != "" { + log.Info("Hostname specified for dhcp lease", "interface", interfaceName, "hostname", instance.DHCPHostname) + client.WithHostName(instance.DHCPHostname) } wg.Go(func() { @@ -666,8 +795,8 @@ func (i *Instance) startDHCP(ctx context.Context, index int, backoffAttempts uin }) // Set the name of the interface so that it can be removed on Service deletion - i.DHCPInterface = interfaceName - i.DHCPInterfaceHwaddr = iface.HardwareAddr.String() + instance.DHCPInterface = interfaceName + instance.DHCPInterfaceHwaddr = iface.HardwareAddr.String() return nil } @@ -747,9 +876,9 @@ func FetchServiceAddresses(s *v1.Service) ([]string, []string) { func FindServiceInstance(svc *v1.Service, instances []*Instance) *Instance { log.Debug("finding service", "namespace", svc.Namespace, "name", svc.Name, "UID", svc.UID) - for i := range instances { - if instances[i].ServiceSnapshot.UID == svc.UID { - return instances[i] + for index := range instances { + if instances[index].UID() == svc.UID { + return instances[index] } } log.Debug("instance not found", "namespace", svc.Namespace, "name", svc.Name, "UID", svc.UID) diff --git a/pkg/instance/instance_test.go b/pkg/instance/instance_test.go new file mode 100644 index 00000000..5e3f5d0a --- /dev/null +++ b/pkg/instance/instance_test.go @@ -0,0 +1,65 @@ +package instance + +import ( + "sync" + "testing" + + "github.com/kube-vip/kube-vip/pkg/kubevip" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +func TestUIDUsesImmutableServiceUID(t *testing.T) { + serviceUID := types.UID("original-service") + instance := &Instance{ + ServiceUID: serviceUID, + ServiceSnapshot: &v1.Service{ObjectMeta: metav1.ObjectMeta{ + UID: types.UID("replacement-service"), + }}, + } + + if got := instance.UID(); got != serviceUID { + t.Fatalf("UID() = %q, want %q", got, serviceUID) + } + + instance.ServiceUID = "" + if got := instance.UID(); got != "" { + t.Fatalf("UID() without ServiceUID = %q, want empty UID", got) + } +} + +func TestCleanupStateIsImmutable(t *testing.T) { + original := &v1.Service{ObjectMeta: metav1.ObjectMeta{ + Name: "service", Namespace: "default", Annotations: map[string]string{kubevip.ServiceLease: "shared"}, + }, Spec: v1.ServiceSpec{LoadBalancerIP: "192.0.2.10", ExternalTrafficPolicy: v1.ServiceExternalTrafficPolicyTypeCluster}} + immutableInfo := serviceCleanupInfo(original) + instance := &Instance{ + ServiceSnapshot: original, + ServiceAddresses: []string{"192.0.2.10"}, + cleanupInfo: &immutableInfo, + } + instance.ServiceSnapshot = &v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "changed"}} + + cleanupInfo, ok := instance.CleanupInfo() + if !ok || cleanupInfo.Namespace != "default" || cleanupInfo.Name != "service" || cleanupInfo.Lease != "shared" || + cleanupInfo.ExternalTrafficPolicy != v1.ServiceExternalTrafficPolicyTypeCluster { + t.Fatalf("CleanupInfo() = %+v, %t, want creation-time Service policy", cleanupInfo, ok) + } +} + +func TestTransferLinkAttachmentOwnershipConcurrent(t *testing.T) { + target := &Instance{IsVLAN: true, VLANInterface: "eth0.42"} + var wg sync.WaitGroup + for range 100 { + wg.Go(func() { + if !transferLinkAttachmentOwnership("eth0.42", []*Instance{target}, true) { + t.Error("transferLinkAttachmentOwnership() did not find target") + } + }) + } + wg.Wait() + if !target.vlanOwned.Load() { + t.Fatal("target did not receive VLAN ownership") + } +} diff --git a/pkg/instance/links_linux_test.go b/pkg/instance/links_linux_test.go new file mode 100644 index 00000000..3f625e78 --- /dev/null +++ b/pkg/instance/links_linux_test.go @@ -0,0 +1,102 @@ +//go:build linux + +package instance + +import ( + "errors" + "os" + "runtime" + "testing" + + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" +) + +// 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 TestCleanupLinkAttachmentsOnlyDeletesOwnedVLAN(t *testing.T) { + for _, test := range []struct { + name string + preexists bool + inUse bool + }{ + {name: "owned VLAN", preexists: false}, + {name: "adopted VLAN", preexists: true}, + {name: "owned VLAN used by another Service", inUse: true}, + } { + t.Run(test.name, func(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) + } + }() + + parent := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kvattach0"}} + if err := netlink.LinkAdd(parent); err != nil { + t.Fatalf("creating parent interface: %v", err) + } + if err := netlink.LinkSetUp(parent); err != nil { + t.Fatalf("bringing up parent interface: %v", err) + } + if test.preexists { + vlan := &netlink.Vlan{LinkAttrs: netlink.LinkAttrs{Name: "kvattach0.42", ParentIndex: parent.Attrs().Index}, VlanId: 42} + if err := netlink.LinkAdd(vlan); err != nil { + t.Fatalf("creating existing VLAN: %v", err) + } + } + + instance := &Instance{} + if err := instance.addVLAN(parent.Attrs().Name, 42); err != nil { + t.Fatalf("adding VLAN attachment: %v", err) + } + if instance.vlanOwned.Load() == test.preexists { + t.Fatalf("vlanOwned = %t, want %t", instance.vlanOwned.Load(), !test.preexists) + } + var remaining []*Instance + if test.inUse { + remaining = []*Instance{{IsVLAN: true, VLANInterface: instance.VLANInterface}} + } + if err := instance.CleanupLinkAttachments(remaining...); err != nil { + t.Fatalf("cleaning attachments: %v", err) + } + if test.inUse { + if !remaining[0].vlanOwned.Load() { + t.Fatal("remaining Service did not receive VLAN cleanup ownership") + } + if _, err := netlink.LinkByName("kvattach0.42"); err != nil { + t.Fatalf("VLAN was removed while a Service still used it: %v", err) + } + if err := remaining[0].CleanupLinkAttachments(); err != nil { + t.Fatalf("cleaning transferred attachment: %v", err) + } + } + + _, err = netlink.LinkByName("kvattach0.42") + var notFound netlink.LinkNotFoundError + if test.preexists && err != nil { + t.Fatalf("adopted VLAN was removed: %v", err) + } + if !test.preexists && !errors.As(err, ¬Found) { + t.Fatalf("owned VLAN remains after cleanup: %v", err) + } + }) + } +} diff --git a/pkg/kubevip/config_validation.go b/pkg/kubevip/config_validation.go index dc36767f..e0af7b70 100644 --- a/pkg/kubevip/config_validation.go +++ b/pkg/kubevip/config_validation.go @@ -2,6 +2,7 @@ package kubevip import ( "fmt" + "math" "net/url" "strings" ) @@ -24,10 +25,23 @@ func (c *Config) Validate() error { if err := validateInstanceName(c.InstanceName); err != nil { return err } + if err := validateRoutingProtocol(c.RoutingProtocol); err != nil { + return err + } return nil } +// validateRoutingProtocol rejects values the kernel cannot represent: netlink +// carries the address and route protocol in a single byte, so a larger value is +// silently truncated on the wire and never matches on readback. +func validateRoutingProtocol(protocol int) error { + if protocol < 0 || protocol > math.MaxUint8 { + return fmt.Errorf("routingProtocol %d is out of range, must be between 0 and %d", protocol, math.MaxUint8) + } + return nil +} + func validateInstanceName(name string) error { if name == "" { return nil diff --git a/pkg/kubevip/config_validation_test.go b/pkg/kubevip/config_validation_test.go index 217e011c..ceadc257 100644 --- a/pkg/kubevip/config_validation_test.go +++ b/pkg/kubevip/config_validation_test.go @@ -62,6 +62,30 @@ func TestValidate_InstanceName(t *testing.T) { } } +func TestValidate_RoutingProtocol(t *testing.T) { + tests := []struct { + name string + protocol int + wantErr bool + }{ + {name: "unset", protocol: 0, wantErr: false}, + {name: "kube-vip default", protocol: 248, wantErr: false}, + {name: "maximum byte value", protocol: 255, wantErr: false}, + {name: "truncated on the wire", protocol: 256, wantErr: true}, + {name: "negative", protocol: -1, wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := &Config{RoutingProtocol: tt.protocol} + err := config.Validate() + if (err != nil) != tt.wantErr { + t.Fatalf("Validate() error = %v, wantErr %t", err, tt.wantErr) + } + }) + } +} + func TestInstanceNameLimitReservesNftablesPrefixAndFamilySuffix(t *testing.T) { name := strings.Repeat("a", instanceNameMaxLength) if got := len(egressNftablesTablePrefix + name + egressNftablesTableSuffix); got != nftablesNameMaxLength { diff --git a/pkg/services/processor_lease_test.go b/pkg/services/processor_lease_test.go index 5188d12e..c39bfe4b 100644 --- a/pkg/services/processor_lease_test.go +++ b/pkg/services/processor_lease_test.go @@ -46,7 +46,7 @@ func TestAddOrModifyStopsTrackedServiceWhenTypeChanges(t *testing.T) { p := &Processor{ config: &kubevip.Config{}, leaseMgr: lease.NewManager(), - ServiceInstances: []*instance.Instance{{ServiceSnapshot: tracked}}, + ServiceInstances: []*instance.Instance{{ServiceUID: uid, ServiceSnapshot: tracked}}, } svcCtx := servicecontext.New(context.Background()) p.svcMap.Store(uid, svcCtx) @@ -191,7 +191,7 @@ func TestOnStoppedLeadingDoesNotDeleteReplacementContext(t *testing.T) { oldCtx := servicecontext.New(context.Background()) replacementCtx := servicecontext.New(context.Background()) p.svcMap.Store(service.UID, replacementCtx) - replacementInstance := &instance.Instance{ServiceSnapshot: service.DeepCopy()} + replacementInstance := &instance.Instance{ServiceUID: service.UID, ServiceSnapshot: service.DeepCopy()} p.ServiceInstances = []*instance.Instance{replacementInstance} leaseNamespace, serviceLease := lease.ServiceName(service) diff --git a/pkg/services/services.go b/pkg/services/services.go index df7bc01c..76e77801 100644 --- a/pkg/services/services.go +++ b/pkg/services/services.go @@ -13,7 +13,6 @@ import ( log "log/slog" "github.com/google/go-cmp/cmp" - "github.com/vishvananda/netlink" v1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -482,36 +481,8 @@ func (p *Processor) deleteService(ctx context.Context, uid types.UID) error { serviceInstance.Clusters[x].Stop() } - if serviceInstance.IsVLAN { - vlan, err := netlink.LinkByName(serviceInstance.VLANInterface) - if err != nil { - return fmt.Errorf("[service] error finding VLAN Interface: %v", err) - } - - err = netlink.LinkDel(vlan) - if err != nil { - return fmt.Errorf("[service] error deleting VLAN interface : %v", err) - } - } - - if serviceInstance.IsDHCPv4 || serviceInstance.IsDHCPv6 { - if serviceInstance.IsDHCPv4 { - serviceInstance.DHCPv4Client.Stop() - } - - if serviceInstance.IsDHCPv6 { - serviceInstance.DHCPv6Client.Stop() - } - - macvlan, err := netlink.LinkByName(serviceInstance.DHCPInterface) - if err != nil { - return fmt.Errorf("[service] error finding VIP Interface: %v", err) - } - - err = netlink.LinkDel(macvlan) - if err != nil { - return fmt.Errorf("[service] error deleting DHCP Link : %v", err) - } + if err := serviceInstance.CleanupLinkAttachments(updatedInstances...); err != nil { + return fmt.Errorf("[service] error cleaning up link attachments: %w", err) } // We will need to tear down the egress diff --git a/pkg/services/services_wireguard.go b/pkg/services/services_wireguard.go index 0081b72c..dbe413dc 100644 --- a/pkg/services/services_wireguard.go +++ b/pkg/services/services_wireguard.go @@ -8,6 +8,7 @@ import ( "github.com/kube-vip/kube-vip/pkg/nftables" "github.com/kube-vip/kube-vip/pkg/utils" + "github.com/kube-vip/kube-vip/pkg/wireguard" v1 "k8s.io/api/core/v1" ) @@ -108,44 +109,49 @@ func (p *Processor) deleteServiceWireguard(_ context.Context, svc *v1.Service) { continue } - portServiceID := fmt.Sprintf("%s_p%d", serviceID, port.Port) - - // Try to delete for both IPv4 and IPv6 if we have mixed IPs - hasIPv4 := false - hasIPv6 := false - for _, vip := range serviceIPs { - // Strip CIDR notation before checking IP version - addr := utils.StripCIDR(vip) - if utils.IsIPv6(addr) { - hasIPv6 = true - } else { - hasIPv4 = true + portServiceIDs := wireguard.ServicePortIDSet(svc.Namespace, svc.Name, port) + for _, portServiceID := range portServiceIDs { + // Try to delete for both IPv4 and IPv6 if we have mixed IPs + hasIPv4 := false + hasIPv6 := false + for _, vip := range serviceIPs { + // Strip CIDR notation before checking IP version + addr := utils.StripCIDR(vip) + if utils.IsIPv6(addr) { + hasIPv6 = true + } else { + hasIPv4 = true + } } - } - if hasIPv4 { - if err := nftables.DeleteIngressChains(false, portServiceID); err != nil { - log.Error("[wireguard] failed to delete IPv4 DNAT chains", - "service", svc.Name, - "port", port.Port, - "err", err) - } else { - log.Debug("[wireguard] deleted IPv4 DNAT chains", - "service", svc.Name, - "port", port.Port) + if hasIPv4 { + if err := nftables.DeleteIngressChains(false, portServiceID); err != nil { + log.Error("[wireguard] failed to delete IPv4 DNAT chains", + "service", svc.Name, + "port", port.Port, + "id", portServiceID, + "err", err) + } else { + log.Debug("[wireguard] deleted IPv4 DNAT chains", + "service", svc.Name, + "port", port.Port, + "id", portServiceID) + } } - } - if hasIPv6 { - if err := nftables.DeleteIngressChains(true, portServiceID); err != nil { - log.Error("[wireguard] failed to delete IPv6 DNAT chains", - "service", svc.Name, - "port", port.Port, - "err", err) - } else { - log.Debug("[wireguard] deleted IPv6 DNAT chains", - "service", svc.Name, - "port", port.Port) + if hasIPv6 { + if err := nftables.DeleteIngressChains(true, portServiceID); err != nil { + log.Error("[wireguard] failed to delete IPv6 DNAT chains", + "service", svc.Name, + "port", port.Port, + "id", portServiceID, + "err", err) + } else { + log.Debug("[wireguard] deleted IPv6 DNAT chains", + "service", svc.Name, + "port", port.Port, + "id", portServiceID) + } } } } diff --git a/pkg/wireguard/service_id.go b/pkg/wireguard/service_id.go new file mode 100644 index 00000000..39858778 --- /dev/null +++ b/pkg/wireguard/service_id.go @@ -0,0 +1,50 @@ +package wireguard + +import ( + "crypto/sha256" + "fmt" + "strings" + + "github.com/kube-vip/kube-vip/pkg/utils" + v1 "k8s.io/api/core/v1" +) + +const maxServicePortIDLength = 50 + +// ServicePortIDs returns the protocol-qualified nftables identifier and the +// prior port-only identifier that must be removed during migration. +// +// Sanitisation maps '-' onto the '_' separator, so "a-b/c" and "a/b-c" would +// otherwise share a chain; a hash of the raw name keeps them distinct. +func ServicePortIDs(namespace, name string, port v1.ServicePort) (string, string) { + rawServiceID := fmt.Sprintf("%s_%s", namespace, name) + serviceID := utils.SanitizeServiceID(rawServiceID) + legacyID := fmt.Sprintf("%s_p%d", serviceID, port.Port) + protocol := port.Protocol + if protocol == "" { + protocol = v1.ProtocolTCP + } + suffix := fmt.Sprintf("_p%d_%s", port.Port, strings.ToLower(string(protocol))) + maxBase := maxServicePortIDLength - len(suffix) + if serviceID != rawServiceID || len(serviceID) > maxBase { + sum := sha256.Sum256([]byte(rawServiceID)) + hash := fmt.Sprintf("_%x", sum[:8]) + maxPrefix := maxBase - len(hash) + if len(serviceID) > maxPrefix { + serviceID = serviceID[:maxPrefix] + } + serviceID += hash + } + return serviceID + suffix, legacyID +} + +// ServicePortIDSet returns the newer protocol-qualified ID first and the legacy +// port-only ID second so callers can actively use the new ID while also +// deleting older rules during migration. +func ServicePortIDSet(namespace, name string, port v1.ServicePort) []string { + currentID, legacyID := ServicePortIDs(namespace, name, port) + if currentID == legacyID { + return []string{currentID} + } + return []string{currentID, legacyID} +} diff --git a/pkg/wireguard/service_id_test.go b/pkg/wireguard/service_id_test.go new file mode 100644 index 00000000..6b95859d --- /dev/null +++ b/pkg/wireguard/service_id_test.go @@ -0,0 +1,44 @@ +package wireguard + +import ( + "strings" + "testing" + + v1 "k8s.io/api/core/v1" +) + +func TestServicePortIDsKeepProtocolAndLongNamesDistinct(t *testing.T) { + udp := v1.ServicePort{Port: 53, Protocol: v1.ProtocolUDP} + udpID, legacyID := ServicePortIDs("default", "dns", udp) + if udpID != "default_dns_p53_udp" || legacyID != "default_dns_p53" { + t.Fatalf("ServicePortIDs() = %q, %q", udpID, legacyID) + } + tcpID, _ := ServicePortIDs("default", "dns", v1.ServicePort{Port: 53, Protocol: v1.ProtocolTCP}) + if tcpID == udpID { + t.Fatal("TCP and UDP Services sharing a port received the same rule ID") + } + + longPrefix := strings.Repeat("a", 63) + first, _ := ServicePortIDs(longPrefix, "first", udp) + second, _ := ServicePortIDs(longPrefix, "second", udp) + if len(first) > maxServicePortIDLength || first == second { + t.Fatalf("long Service IDs = %q, %q", first, second) + } + + hyphenatedNamespace, _ := ServicePortIDs("a-b", "c", udp) + hyphenatedName, _ := ServicePortIDs("a", "b-c", udp) + if hyphenatedNamespace == hyphenatedName { + t.Fatalf("distinct Services received the same rule ID %q", hyphenatedNamespace) + } +} + +func TestServicePortIDSetIncludesLegacyCleanupID(t *testing.T) { + udp := v1.ServicePort{Port: 53, Protocol: v1.ProtocolUDP} + ids := ServicePortIDSet("default", "dns", udp) + if len(ids) != 2 { + t.Fatalf("ServicePortIDSet() length = %d, want 2", len(ids)) + } + if ids[0] != "default_dns_p53_udp" || ids[1] != "default_dns_p53" { + t.Fatalf("ServicePortIDSet() = %q, want [%q %q]", ids, "default_dns_p53_udp", "default_dns_p53") + } +} diff --git a/testing/e2e/e2e_test.go b/testing/e2e/e2e_test.go index e5a56655..5fd3ca8d 100644 --- a/testing/e2e/e2e_test.go +++ b/testing/e2e/e2e_test.go @@ -1288,7 +1288,8 @@ func killLeader(leaderName string) { session, err := gexec.Start(cmd, GinkgoWriter, GinkgoWriter) Expect(err).NotTo(HaveOccurred()) - Eventually(session, "5s").Should(gexec.Exit(0)) + // Every parallel kind cluster shares one Docker daemon, so the kill can be acknowledged slowly. + Eventually(session, "30s").Should(gexec.Exit(0)) } func findLeader(leaderIPAddr string, clusterName string) string {