refactor(endpoints): reconcile every endpoint event through one path

Now that a deletion recomputes the remaining endpoints, AddOrModify and
Delete differed only in how the event mutated the provider. Keeping two
near-identical bodies is what let their behaviour drift apart in the first
place, so fold them into a single Reconcile that dispatches the mutation
in applyEvent.

Deletions consequently pick up the parts of the add path they were still
missing: the IPv6 egress guard, the leader election start, and the
endpointless opt-in branch, which now keeps an opted-in Cluster service up
after its last EndpointSlice disappears instead of tearing it down.

Assisted-by: GitHub-Copilot:unspecified
Signed-off-by: Marcel Fest <marcel.fest@telekom.de>

CHANGELOG.md

@@ -50,6 +50,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Added support in ipoib interfaces in ARP mode. Fixes #694
 
 ### Changed
+- BGP mode now honours `enable_leader_election` for services: a single global services leader advertises the service VIPs instead of every node advertising them. Deployments that enabled `enable_leader_election` for the control plane and relied on ECMP/multipath for services must unset it (or switch to `enable_service_election`) to keep the previous datapath. kube-vip logs a warning on startup when this path is taken.
 - Updated signal handlers in manager_arp.go, manager_bgp.go, manager_wireguard.go, and manager_table.go to use switch statement pattern for handling multiple signals (SIGUSR1, SIGINT, SIGTERM)
 - wireguard.go now manages a complete wireguard interface on the current network namespace
 - manager_wireguard.go uses the new wireguard.go implementation

pkg/endpoints/endpoints.go

@@ -46,20 +46,24 @@ func NewEndpointProcessor(config *kubevip.Config, provider providers.Provider, b
 	}
 }
 
-func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Event,
+// Reconcile applies a watch event to the provider and reconciles the service
+// against the endpoints that remain afterwards. A deleted object is only one of
+// potentially several backing the service, so deletions are recomputed rather
+// than assumed to empty it. It reports whether the caller should skip this event
+// and wait for the next one.
+func (p *Processor) Reconcile(svcCtx *servicecontext.Context, event watch.Event,
 	lastKnownGoodEndpoint *string, service *v1.Service, id string,
 	serviceFunc func(*servicecontext.Context, *v1.Service, *sync.WaitGroup, bool) error, wg *sync.WaitGroup,
 	clientSet *kubernetes.Clientset,
 	egressUpdateFunc func(context.Context, *v1.Service) error) (bool, error) {
 
-	var err error
-	if err = p.provider.LoadObject(event.Object, svcCtx.Cancel); err != nil {
-		return false, fmt.Errorf("[%s] error loading k8s object: %w", p.provider.GetLabel(), err)
+	if err := p.applyEvent(svcCtx, event); err != nil {
+		return false, err
 	}
 
 	endpoints, err := p.worker.getEndpoints(service, id)
 	if err != nil {
-		return false, err
+		return false, fmt.Errorf("[%s] error getting endpoints: %w", p.provider.GetLabel(), err)
 	}
 
 	if err := p.worker.setInstanceEndpointsStatus(svcCtx.Ctx, service, endpoints); err != nil {
@@ -89,11 +93,7 @@ func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Even
 
 		svcCtx.SignalReadiness()
 
-		// There are local endpoints available on the node
-		// Process immediately if:
-		// - No services/leader election is enabled, OR
-		// - WireGuard is enabled (it always needs immediate DNAT rule updates)
-		if (!p.config.EnableServicesElection && !p.config.EnableLeaderElection) || p.config.EnableWireguard {
+		if p.shouldProcessInstance() {
 			if err := p.worker.processInstance(svcCtx, service); err != nil {
 				return false, fmt.Errorf("failed to process non-empty instance: %w", err)
 			}
@@ -106,21 +106,13 @@ func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Even
 			}
 			svcCtx.SignalReadiness()
 
-			if (!p.config.EnableServicesElection && !p.config.EnableLeaderElection) || p.config.EnableWireguard {
+			if p.shouldProcessInstance() {
 				if err := p.worker.processInstance(svcCtx, service); err != nil {
 					return false, fmt.Errorf("failed to process endpointless instance: %w", err)
 				}
 			}
 		} else if svcCtx.Signalled.Load() {
-			// There are no local endpoints
-			svcCtx.ResetReadiness()
-			p.worker.clear(svcCtx, lastKnownGoodEndpoint, service)
-			if p.config.EnableARP && !p.config.EnableServicesElection {
-				i := instance.FindServiceInstance(service, *p.instances)
-				for _, c := range i.Clusters {
-					c.Stop()
-				}
-			}
+			p.handleNoEndpoints(svcCtx, service, lastKnownGoodEndpoint)
 		}
 	}
 
@@ -133,13 +125,42 @@ func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Even
 	return false, nil
 }
 
-func (p *Processor) Delete(ctx context.Context, service *v1.Service, id string) error {
-	if err := p.worker.delete(ctx, service, id); err != nil {
-		return fmt.Errorf("[%s] error deleting service: %w", p.provider.GetLabel(), err)
+// applyEvent updates the provider's view of the objects backing this service.
+func (p *Processor) applyEvent(svcCtx *servicecontext.Context, event watch.Event) error {
+	if event.Type == watch.Deleted {
+		if err := p.provider.DeleteObject(event.Object); err != nil {
+			return fmt.Errorf("[%s] error deleting k8s object: %w", p.provider.GetLabel(), err)
+		}
+		return nil
+	}
+
+	if err := p.provider.LoadObject(event.Object, svcCtx.Cancel); err != nil {
+		return fmt.Errorf("[%s] error loading k8s object: %w", p.provider.GetLabel(), err)
 	}
 	return nil
 }
 
+// shouldProcessInstance reports whether this node has to program the datapath
+// itself, rather than waiting to be told to by a leader election callback.
+// WireGuard always reprograms, because its DNAT rules are per-endpoint.
+func (p *Processor) shouldProcessInstance() bool {
+	return (!p.config.EnableServicesElection && !p.config.EnableLeaderElection) || p.config.EnableWireguard
+}
+
+// handleNoEndpoints tears down everything backing a service that no longer has
+// any usable endpoints.
+func (p *Processor) handleNoEndpoints(svcCtx *servicecontext.Context, service *v1.Service, lastKnownGoodEndpoint *string) {
+	svcCtx.ResetReadiness()
+	p.worker.clear(svcCtx, lastKnownGoodEndpoint, service)
+	if p.config.EnableARP && !p.config.EnableServicesElection && p.instances != nil {
+		if i := instance.FindServiceInstance(service, *p.instances); i != nil {
+			for _, c := range i.Clusters {
+				c.Stop()
+			}
+		}
+	}
+}
+
 func (p *Processor) updateLastKnownGoodEndpoint(lastKnownGoodEndpoint *string, endpoints []string, service *v1.Service) {
 	// if we haven't populated one, then do so
 	family := utils.IPv4Family

pkg/endpoints/endpoints_bgp.go

@@ -2,7 +2,6 @@ package endpoints
 
 import (
 	"context"
-	"fmt"
 	log "log/slog"
 
 	"github.com/kube-vip/kube-vip/pkg/bgp"
@@ -76,32 +75,6 @@ func (b *BGP) getEndpoints(service *v1.Service, id string) ([]string, error) {
 	return b.getAllEndpoints(service, id)
 }
 
-func (b *BGP) delete(ctx context.Context, service *v1.Service, id string) error {
-	// When no-leader-elecition mode
-	if !b.config.EnableServicesElection && !b.config.EnableLeaderElection {
-		// find all existing local endpoints
-		endpoints, err := b.getEndpoints(service, id)
-		if err != nil {
-			return fmt.Errorf("[%s] error getting endpoints: %w", b.provider.GetLabel(), err)
-		}
-
-		// If there were local endpoints deleted
-		if len(endpoints) > 0 {
-			b.deleteAction(ctx, service)
-		}
-	}
-
-	return nil
-}
-
-func (b *BGP) deleteAction(ctx context.Context, service *v1.Service) {
-	b.clearBGPHosts(ctx, service)
-}
-
-func (b *BGP) clearBGPHosts(ctx context.Context, service *v1.Service) {
-	ClearBGPHosts(ctx, service, b.instances, b.bgpServer)
-}
-
 func (b *BGP) setInstanceEndpointsStatus(_ context.Context, _ *v1.Service, _ []string) error {
 	return nil
 }

pkg/endpoints/endpoints_generic.go

@@ -22,7 +22,6 @@ type endpointWorker interface {
 	clear(svcCtx *servicecontext.Context, lastKnownGoodEndpoint *string, service *v1.Service)
 	getEndpoints(service *v1.Service, id string) ([]string, error)
 	removeEgress(service *v1.Service, lastKnownGoodEndpoint *string)
-	delete(ctx context.Context, service *v1.Service, id string) error
 	setInstanceEndpointsStatus(ctx context.Context, service *v1.Service, endpoints []string) error
 }
 
@@ -105,10 +104,6 @@ func (g *generic) getAllEndpoints(service *v1.Service, id string) ([]string, err
 func (g *generic) removeEgress(_ *v1.Service, _ *string) {
 }
 
-func (g *generic) delete(_ context.Context, _ *v1.Service, _ string) error {
-	return nil
-}
-
 func (g *generic) setInstanceEndpointsStatus(_ context.Context, _ *v1.Service, _ []string) error {
 	return nil
 }

pkg/endpoints/endpoints_routing_table.go

@@ -82,28 +82,6 @@ func (rt *RoutingTable) removeEgress(service *v1.Service, lastKnownGoodEndpoint
 	}
 }
 
-func (rt *RoutingTable) delete(_ context.Context, service *v1.Service, id string) error {
-	// When no-leader-elecition mode
-	if !rt.config.EnableServicesElection && !rt.config.EnableLeaderElection {
-		// find all existing local endpoints
-		endpoints, err := rt.getEndpoints(service, id)
-		if err != nil {
-			return fmt.Errorf("[%s] error getting endpoints: %w", rt.provider.GetLabel(), err)
-		}
-
-		// If there were local endpoints deleted
-		if len(endpoints) > 0 {
-			rt.deleteAction(service)
-		}
-	}
-
-	return nil
-}
-
-func (rt *RoutingTable) deleteAction(service *v1.Service) {
-	ClearRoutes(service, rt.instances, rt.routeMgr)
-}
-
 func (rt *RoutingTable) setInstanceEndpointsStatus(ctx context.Context, service *v1.Service, endpoints []string) error {
 	inst := instance.FindServiceInstance(service, *rt.instances)
 	if inst == nil {

pkg/endpoints/endpoints_test.go

@@ -58,14 +58,101 @@ func (f *fakeWorker) clear(_ *servicecontext.Context, _ *string, _ *v1.Service)
 
 func (f *fakeWorker) getEndpoints(_ *v1.Service, _ string) ([]string, error) { return f.endpoints, nil }
 func (f *fakeWorker) removeEgress(_ *v1.Service, _ *string)                  {}
-func (f *fakeWorker) delete(_ context.Context, _ *v1.Service, _ string) error {
-	return nil
-}
 func (f *fakeWorker) setInstanceEndpointsStatus(_ context.Context, _ *v1.Service, _ []string) error {
 	return nil
 }
 
-func TestAddOrModify_ZeroEndpointsBehavior(t *testing.T) {
+// TestReconcile_RecomputesRemainingEndpoints asserts that deleting one EndpointSlice
+// reconciles against the endpoints that remain, instead of assuming the service
+// lost all of them.
+func TestReconcile_RecomputesRemainingEndpoints(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name              string
+		remaining         []string
+		lastKnown         string
+		expectReady       bool
+		expectClear       bool
+		expectProcess     bool
+		expectedLastKnown string
+	}{
+		{
+			name:              "remaining endpoints keep the service up",
+			remaining:         []string{"10.0.0.2"},
+			lastKnown:         "10.0.0.2",
+			expectReady:       true,
+			expectProcess:     true,
+			expectedLastKnown: "10.0.0.2",
+		},
+		{
+			name:              "stale last known endpoint moves to a survivor",
+			remaining:         []string{"10.0.0.2"},
+			lastKnown:         "10.0.0.1",
+			expectReady:       true,
+			expectProcess:     true,
+			expectedLastKnown: "10.0.0.2",
+		},
+		{
+			name:        "last endpoint removed tears the service down",
+			remaining:   nil,
+			lastKnown:   "10.0.0.1",
+			expectReady: false,
+			expectClear: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			worker := &fakeWorker{endpoints: test.remaining}
+			p := &Processor{
+				config:   &kubevip.Config{},
+				provider: providers.NewEndpointslices(),
+				worker:   worker,
+			}
+
+			svcCtx := servicecontext.New(context.Background())
+			svcCtx.SignalReadiness()
+
+			lastKnown := test.lastKnown
+			restart, err := p.Reconcile(
+				svcCtx,
+				watch.Event{
+					Type:   watch.Deleted,
+					Object: &discoveryv1.EndpointSlice{ObjectMeta: metav1.ObjectMeta{Name: "slice-1"}},
+				},
+				&lastKnown,
+				&v1.Service{Spec: v1.ServiceSpec{ExternalTrafficPolicy: v1.ServiceExternalTrafficPolicyTypeLocal}},
+				"node-1",
+				func(*servicecontext.Context, *v1.Service, *sync.WaitGroup, bool) error { return nil },
+				&sync.WaitGroup{},
+				nil,
+				nil,
+			)
+			if err != nil {
+				t.Fatalf("Reconcile returned error: %v", err)
+			}
+			if restart {
+				t.Fatal("Reconcile unexpectedly requested restart")
+			}
+
+			if ready := svcCtx.Signalled.Load(); ready != test.expectReady {
+				t.Fatalf("readiness mismatch: expected %v, got %v", test.expectReady, ready)
+			}
+			if worker.clearCalled != test.expectClear {
+				t.Fatalf("clearCalled mismatch: expected %v, got %v", test.expectClear, worker.clearCalled)
+			}
+			if worker.processCalled != test.expectProcess {
+				t.Fatalf("processCalled mismatch: expected %v, got %v", test.expectProcess, worker.processCalled)
+			}
+			if test.expectedLastKnown != "" && lastKnown != test.expectedLastKnown {
+				t.Fatalf("lastKnownGoodEndpoint mismatch: expected %q, got %q", test.expectedLastKnown, lastKnown)
+			}
+		})
+	}
+}
+
+func TestReconcile_ZeroEndpointsBehavior(t *testing.T) {
 	t.Parallel()
 
 	run := func(t *testing.T, service *v1.Service, presetSignalled bool, expectReady bool, expectClear bool, expectProcess bool) {
@@ -83,7 +170,7 @@ func TestAddOrModify_ZeroEndpointsBehavior(t *testing.T) {
 			svcCtx.SignalReadiness()
 		}
 
-		restart, err := p.AddOrModify(
+		restart, err := p.Reconcile(
 			svcCtx,
 			watch.Event{Type: watch.Modified, Object: &discoveryv1.EndpointSlice{}},
 			new(string),
@@ -95,10 +182,10 @@ func TestAddOrModify_ZeroEndpointsBehavior(t *testing.T) {
 			nil,
 		)
 		if err != nil {
-			t.Fatalf("AddOrModify returned error: %v", err)
+			t.Fatalf("Reconcile returned error: %v", err)
 		}
 		if restart {
-			t.Fatal("AddOrModify unexpectedly requested restart")
+			t.Fatal("Reconcile unexpectedly requested restart")
 		}
 
 		if ready := svcCtx.Signalled.Load(); ready != expectReady {
@@ -137,15 +224,15 @@ func TestAddOrModify_ZeroEndpointsBehavior(t *testing.T) {
 	})
 }
 
-// TestAddOrModify_ServicesElectionStartsOnce asserts that repeated endpoint events
+// TestReconcile_ServicesElectionStartsOnce asserts that repeated endpoint events
 // for the same service start the leader-election restart loop exactly once.
 //
-// AddOrModify runs on every EndpointSlice add/modify/resync event, and the loop it
+// Reconcile runs on every EndpointSlice add/modify/resync event, and the loop it
 // starts only returns once the service context is cancelled. Starting it per event
 // therefore accumulates duplicate goroutines that all contend on the same lease.
 //
 // See https://github.com/kube-vip/kube-vip/issues/1665.
-func TestAddOrModify_ServicesElectionStartsOnce(t *testing.T) {
+func TestReconcile_ServicesElectionStartsOnce(t *testing.T) {
 	config := &kubevip.Config{
 		EnableServicesElection: true,
 		LeaderElectionType:     "kubernetes",
@@ -189,13 +276,13 @@ func TestAddOrModify_ServicesElectionStartsOnce(t *testing.T) {
 
 	// Three endpoint events, as a flapping backend pod would produce.
 	for range 3 {
-		restart, err := p.AddOrModify(svcCtx, watch.Event{Type: watch.Modified, Object: &discoveryv1.EndpointSlice{}},
+		restart, err := p.Reconcile(svcCtx, watch.Event{Type: watch.Modified, Object: &discoveryv1.EndpointSlice{}},
 			new(string), service, "node-1", serviceFunc, wg, nil, nil)
 		if err != nil {
-			t.Fatalf("AddOrModify returned error: %v", err)
+			t.Fatalf("Reconcile returned error: %v", err)
 		}
 		if restart {
-			t.Fatal("AddOrModify unexpectedly requested restart")
+			t.Fatal("Reconcile unexpectedly requested restart")
 		}
 	}
 

pkg/endpoints/endpoints_wireguard.go

@@ -242,14 +242,6 @@ func (w *wireguardWorker) removeEgress(service *v1.Service, lastKnownGoodEndpoin
 	log.Debug("[wireguard] removeEgress called (no-op)", "service", service.Name)
 }
 
-// delete removes all DNAT rules for a service
-func (w *wireguardWorker) delete(ctx context.Context, service *v1.Service, id string) error {
-	log.Info("[wireguard] deleting DNAT rules for service", "service", service.Name, "namespace", service.Namespace)
-
-	w.clear(nil, nil, service)
-	return nil
-}
-
 // setInstanceEndpointsStatus updates the endpoint status on the service instance
 func (w *wireguardWorker) setInstanceEndpointsStatus(_ context.Context, service *v1.Service, endpoints []string) error {
 	hasEndpoints := len(endpoints) > 0

pkg/endpoints/providers/endpoints.go

@@ -61,6 +61,18 @@ func (ep *Endpoints) LoadObject(endpoints runtime.Object, cancel context.CancelF
 	return nil
 }
 
+// DeleteObject drops the tracked object. A service is backed by exactly one
+// v1.Endpoints object, so there is nothing to match on and the cache is reset.
+func (ep *Endpoints) DeleteObject(endpoints runtime.Object) error {
+	//nolint:staticcheck // SA1019 endpoints have to be explicitly requested now
+	if _, ok := endpoints.(*v1.Endpoints); !ok {
+		return fmt.Errorf("[%s] unable to parse Kubernetes object", ep.GetLabel())
+	}
+	//nolint:staticcheck // SA1019 endpoints have to be explicitly requested now
+	ep.endpoints = &v1.Endpoints{}
+	return nil
+}
+
 func (ep *Endpoints) GetAllEndpoints() ([]string, error) {
 	result := []string{}
 	for subset := range ep.endpoints.Subsets {

pkg/endpoints/providers/endpointslices.go

@@ -20,15 +20,14 @@ import (
 )
 
 type Endpointslices struct {
-	label       string
-	endpointsv4 []discoveryv1.Endpoint
-	endpointsv6 []discoveryv1.Endpoint
-	ports       []discoveryv1.EndpointPort
+	label  string
+	slices map[string]*discoveryv1.EndpointSlice
 }
 
 func NewEndpointslices() Provider {
 	return &Endpointslices{
-		label: "endpointslices",
+		label:  "endpointslices",
+		slices: make(map[string]*discoveryv1.EndpointSlice),
 	}
 }
 
@@ -59,56 +58,71 @@ func (ep *Endpointslices) LoadObject(endpoints runtime.Object, cancel context.Ca
 		return fmt.Errorf("[%s] error casting endpoints to v1.Endpoints struct", ep.label)
 	}
 
-	if eps.AddressType == discoveryv1.AddressTypeIPv6 {
-		ep.endpointsv6 = eps.Endpoints
-	} else {
-		ep.endpointsv4 = eps.Endpoints
+	if ep.slices == nil {
+		ep.slices = make(map[string]*discoveryv1.EndpointSlice)
 	}
-
-	// Store ports for resolving named ports
-	ep.ports = eps.Ports
+	ep.slices[eps.Name] = eps.DeepCopy()
 
 	return nil
 }
 
+func (ep *Endpointslices) DeleteObject(endpoints runtime.Object) error {
+	eps, ok := endpoints.(*discoveryv1.EndpointSlice)
+	if !ok {
+		return fmt.Errorf("[%s] unable to parse Kubernetes object", ep.GetLabel())
+	}
+	delete(ep.slices, eps.Name)
+	return nil
+}
+
+// isServing reports whether an endpoint should receive traffic. Per the
+// EndpointConditions godoc a nil Serving defers to Ready, and a nil Ready is an
+// unknown state that consumers should interpret as ready.
+func isServing(conditions discoveryv1.EndpointConditions) bool {
+	serving := conditions.Serving
+	if serving == nil {
+		serving = conditions.Ready
+	}
+	return serving == nil || *serving
+}
+
 func (ep *Endpointslices) GetAllEndpoints() ([]string, error) {
 	result := []string{}
-	for _, e := range ep.endpointsv4 {
-		result = append(result, e.Addresses...)
-	}
-	for _, e := range ep.endpointsv6 {
-		result = append(result, e.Addresses...)
+	for _, eps := range ep.slices {
+		for _, e := range eps.Endpoints {
+			if !isServing(e.Conditions) {
+				continue
+			}
+			result = append(result, e.Addresses...)
+		}
 	}
 	return result, nil
 }
 
 func (ep *Endpointslices) GetLocalEndpoints(id string, _ *kubevip.Config) ([]string, error) {
 	var localEndpoints []string
-	tmpEps := []discoveryv1.Endpoint{}
-
-	tmpEps = append(tmpEps, ep.endpointsv4...)
-	tmpEps = append(tmpEps, ep.endpointsv6...)
-
-	for _, endpoint := range tmpEps {
-		if endpoint.Conditions.Serving == nil || !*endpoint.Conditions.Serving {
-			continue
-		}
-		for _, address := range endpoint.Addresses {
-			// 1. Compare the Nodename
-			if endpoint.NodeName != nil && id == *endpoint.NodeName {
-				if endpoint.Hostname != nil {
-					log.Debug("found endpoint", "provider", ep.label, "ip", address, "hostname", *endpoint.Hostname, "nodename", *endpoint.NodeName)
-				} else {
-					log.Debug("found endpoint", "provider", ep.label, "ip", address, "nodename", *endpoint.NodeName)
-				}
-				localEndpoints = append(localEndpoints, address)
+	for _, eps := range ep.slices {
+		for _, endpoint := range eps.Endpoints {
+			if !isServing(endpoint.Conditions) {
 				continue
 			}
+			for _, address := range endpoint.Addresses {
+				// 1. Compare the Nodename
+				if endpoint.NodeName != nil && id == *endpoint.NodeName {
+					if endpoint.Hostname != nil {
+						log.Debug("found endpoint", "provider", ep.label, "ip", address, "hostname", *endpoint.Hostname, "nodename", *endpoint.NodeName)
+					} else {
+						log.Debug("found endpoint", "provider", ep.label, "ip", address, "nodename", *endpoint.NodeName)
+					}
+					localEndpoints = append(localEndpoints, address)
+					continue
+				}
 
-			// 2. Compare the Hostname (only useful if endpoint.NodeName is not available)
-			if endpoint.Hostname != nil && id == *endpoint.Hostname {
-				log.Debug("found endpoint", "provider", ep.label, "ip", address, "hostname", *endpoint.Hostname)
-				localEndpoints = append(localEndpoints, address)
+				// 2. Compare the Hostname (only useful if endpoint.NodeName is not available)
+				if endpoint.Hostname != nil && id == *endpoint.Hostname {
+					log.Debug("found endpoint", "provider", ep.label, "ip", address, "hostname", *endpoint.Hostname)
+					localEndpoints = append(localEndpoints, address)
+				}
 			}
 		}
 	}
@@ -153,9 +167,11 @@ func (ep *Endpointslices) GetLabel() string {
 
 func (ep *Endpointslices) ResolvePort(servicePort v1.ServicePort) int32 {
 	return ResolvePortWithLookup(servicePort, func(name string) int32 {
-		for _, p := range ep.ports {
-			if p.Name != nil && *p.Name == name && p.Port != nil {
-				return *p.Port
+		for _, eps := range ep.slices {
+			for _, p := range eps.Ports {
+				if p.Name != nil && *p.Name == name && p.Port != nil {
+					return *p.Port
+				}
 			}
 		}
 		return 0

pkg/endpoints/providers/endpointslices_test.go

@@ -0,0 +1,149 @@
+package providers
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kube-vip/kube-vip/pkg/kubevip"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestEndpointslicesTracksAndDeletesSlices(t *testing.T) {
+	provider := NewEndpointslices().(*Endpointslices)
+	serving := true
+	nodeName := "node-1"
+
+	slice1 := &discoveryv1.EndpointSlice{
+		ObjectMeta:  metav1.ObjectMeta{Name: "slice-1"},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints: []discoveryv1.Endpoint{{
+			Addresses:  []string{"10.0.0.1"},
+			Conditions: discoveryv1.EndpointConditions{Serving: &serving},
+			NodeName:   &nodeName,
+		}},
+	}
+	slice2 := &discoveryv1.EndpointSlice{
+		ObjectMeta:  metav1.ObjectMeta{Name: "slice-2"},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints: []discoveryv1.Endpoint{{
+			Addresses:  []string{"10.0.0.2"},
+			Conditions: discoveryv1.EndpointConditions{Serving: &serving},
+			NodeName:   &nodeName,
+		}},
+	}
+
+	for _, slice := range []*discoveryv1.EndpointSlice{slice1, slice2} {
+		if err := provider.LoadObject(slice, func() {}); err != nil {
+			t.Fatalf("LoadObject returned error: %v", err)
+		}
+	}
+
+	assertEndpoints(t, provider, []string{"10.0.0.1", "10.0.0.2"})
+	assertLocalEndpoints(t, provider, nodeName, []string{"10.0.0.1", "10.0.0.2"})
+
+	if err := provider.DeleteObject(slice1); err != nil {
+		t.Fatalf("DeleteObject returned error: %v", err)
+	}
+	assertEndpoints(t, provider, []string{"10.0.0.2"})
+	assertLocalEndpoints(t, provider, nodeName, []string{"10.0.0.2"})
+
+	if err := provider.DeleteObject(slice2); err != nil {
+		t.Fatalf("DeleteObject returned error: %v", err)
+	}
+	assertEndpoints(t, provider, nil)
+	assertLocalEndpoints(t, provider, nodeName, nil)
+}
+
+func TestEndpointslicesReplacingSliceUpdatesState(t *testing.T) {
+	provider := NewEndpointslices().(*Endpointslices)
+	first := &discoveryv1.EndpointSlice{
+		ObjectMeta:  metav1.ObjectMeta{Name: "slice-1"},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints:   []discoveryv1.Endpoint{{Addresses: []string{"10.0.0.1"}}},
+	}
+	replacement := first.DeepCopy()
+	replacement.Endpoints[0].Addresses = []string{"10.0.0.2"}
+
+	if err := provider.LoadObject(first, context.CancelFunc(func() {})); err != nil {
+		t.Fatalf("LoadObject returned error: %v", err)
+	}
+	if err := provider.LoadObject(replacement, context.CancelFunc(func() {})); err != nil {
+		t.Fatalf("LoadObject returned error: %v", err)
+	}
+
+	assertEndpoints(t, provider, []string{"10.0.0.2"})
+}
+
+func TestEndpointslicesEndpointConditions(t *testing.T) {
+	yes, no := true, false
+	nodeName := "node-1"
+
+	tests := []struct {
+		name       string
+		conditions discoveryv1.EndpointConditions
+		want       []string
+	}{
+		{"serving true", discoveryv1.EndpointConditions{Serving: &yes}, []string{"10.0.0.1"}},
+		{"serving false", discoveryv1.EndpointConditions{Serving: &no}, nil},
+		{"serving false overrides ready true", discoveryv1.EndpointConditions{Serving: &no, Ready: &yes}, nil},
+		{"nil serving defers to ready true", discoveryv1.EndpointConditions{Ready: &yes}, []string{"10.0.0.1"}},
+		{"nil serving defers to ready false", discoveryv1.EndpointConditions{Ready: &no}, nil},
+		{"both nil is treated as ready", discoveryv1.EndpointConditions{}, []string{"10.0.0.1"}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			provider := NewEndpointslices().(*Endpointslices)
+			slice := &discoveryv1.EndpointSlice{
+				ObjectMeta:  metav1.ObjectMeta{Name: "slice-1"},
+				AddressType: discoveryv1.AddressTypeIPv4,
+				Endpoints: []discoveryv1.Endpoint{{
+					Addresses:  []string{"10.0.0.1"},
+					Conditions: test.conditions,
+					NodeName:   &nodeName,
+				}},
+			}
+			if err := provider.LoadObject(slice, func() {}); err != nil {
+				t.Fatalf("LoadObject returned error: %v", err)
+			}
+			// Cluster and Local policy have to agree on which endpoints are usable.
+			assertEndpoints(t, provider, test.want)
+			assertLocalEndpoints(t, provider, nodeName, test.want)
+		})
+	}
+}
+
+func assertEndpoints(t *testing.T, provider *Endpointslices, want []string) {
+	t.Helper()
+	got, err := provider.GetAllEndpoints()
+	if err != nil {
+		t.Fatalf("GetAllEndpoints returned error: %v", err)
+	}
+	assertStringSet(t, got, want)
+}
+
+func assertLocalEndpoints(t *testing.T, provider *Endpointslices, nodeName string, want []string) {
+	t.Helper()
+	got, err := provider.GetLocalEndpoints(nodeName, &kubevip.Config{})
+	if err != nil {
+		t.Fatalf("GetLocalEndpoints returned error: %v", err)
+	}
+	assertStringSet(t, got, want)
+}
+
+func assertStringSet(t *testing.T, got, want []string) {
+	t.Helper()
+	counts := map[string]int{}
+	for _, value := range got {
+		counts[value]++
+	}
+	for _, value := range want {
+		counts[value]--
+	}
+	for value, count := range counts {
+		if count != 0 {
+			t.Fatalf("endpoint set mismatch for %q: got %v, want %v", value, got, want)
+		}
+	}
+}

pkg/endpoints/providers/interface.go

@@ -18,6 +18,7 @@ type Provider interface {
 	GetLabel() string
 	UpdateServiceAnnotation(context.Context, string, string, *v1.Service, *kubernetes.Clientset) error
 	LoadObject(runtime.Object, context.CancelFunc) error
+	DeleteObject(runtime.Object) error
 	// ResolvePort resolves a service port to the actual target port.
 	// For named ports, it looks up the port number from the endpoint.
 	// For numeric ports, it returns the port as-is.

pkg/manager/worker/bgp.go

@@ -113,6 +113,10 @@ func (b *BGP) StartServices(ctx context.Context) error {
 		if err := b.PerServiceLeader(ctx, false); err != nil {
 			return err
 		}
+	} else if b.config.EnableLeaderElection {
+		log.Warn("leader election is enabled, only the elected leader will advertise service VIPs; unset enable_leader_election to keep advertising from every node (ECMP)",
+			"lease", b.config.ServicesLeaseName)
+		b.GlobalLeader(ctx, b.config.ServicesLeaseName)
 	} else {
 		if err := b.ServicesNoLeader(ctx); err != nil {
 			return err
@@ -121,10 +125,6 @@ func (b *BGP) StartServices(ctx context.Context) error {
 	return nil
 }
 
-func (b *BGP) ServicesGlobalLeader(ctx context.Context, id string) {
-	// NOT IMPLEMENTED
-}
-
 func (b *BGP) Name() string {
 	return "BGP"
 }

pkg/manager/worker/wireguard.go

@@ -102,12 +102,15 @@ func (w *WireGuard) ConfigureServices() {
 }
 
 func (w *WireGuard) StartServices(ctx context.Context) error {
+	// WireGuard has no multipath mechanism, so every service must be advertised by
+	// exactly one node: leader election (per-service or global) is required.
 	if w.config.EnableServicesElection {
 		log.Info("beginning watching services, leaderelection will happen for every service")
-		err := w.svcProcessor.StartServicesWatchForLeaderElection(ctx, false)
-		if err != nil {
+		if err := w.svcProcessor.StartServicesWatchForLeaderElection(ctx, false); err != nil {
 			return err
 		}
+	} else {
+		w.GlobalLeader(ctx, w.config.ServicesLeaseName)
 	}
 	return nil
 }
@@ -147,12 +150,6 @@ func (w *WireGuard) OnStartedLeading(ctx context.Context) {
 	w.endpointWatcherWg.Go(func() {
 		w.watchKubernetesEndpoints(w.endpointWatcherCtx, tunnelConfig)
 	})
-
-	if w.config.EnableServices && !w.config.EnableServicesElection {
-		if err := w.svcProcessor.ServicesWatcher(ctx, services.NewCallback(w.svcProcessor.SyncServices, false), false); err != nil {
-			log.Error("failed to start services watcher", "err", err)
-		}
-	}
 }
 
 // watchKubernetesEndpoints watches the kubernetes service EndpointSlices for changes
@@ -185,8 +182,15 @@ func (w *WireGuard) watchKubernetesEndpoints(ctx context.Context, tunnelConfig *
 
 		switch event.Type {
 		case watch.Added, watch.Modified, watch.Deleted:
-			if err := provider.LoadObject(event.Object, func() {}); err != nil {
-				log.Error("failed to load endpoint object", "err", err)
+			// A deleted slice has to be dropped so it stops counting toward the endpoint set.
+			var err error
+			if event.Type == watch.Deleted {
+				err = provider.DeleteObject(event.Object)
+			} else {
+				err = provider.LoadObject(event.Object, func() {})
+			}
+			if err != nil {
+				log.Error("failed to update endpoint object", "eventType", event.Type, "err", err)
 				continue
 			}
 			endpoints, _ := provider.GetAllEndpoints()

pkg/services/services.go

@@ -463,7 +463,9 @@ func (p *Processor) deleteService(ctx context.Context, uid types.UID) error {
 		endpoints.ClearBGPHostsByInstance(ctx, serviceInstance, p.bgpServer)
 	}
 
-	if p.config.EnableRoutingTable && (p.config.EnableLeaderElection || p.config.EnableServicesElection) {
+	// ClearRoutesByInstance is reference-counted per route, so calling it here is safe
+	// even when the no-election path in Processor.Delete already cleared it.
+	if p.config.EnableRoutingTable {
 		if errs := endpoints.ClearRoutesByInstance(serviceInstance.ServiceSnapshot, serviceInstance, &p.ServiceInstances, p.routeMgr); len(errs) > 0 {
 			for _, err := range errs {
 				log.Error("unable to clear routes", "err", err)

pkg/services/watch_endpoints.go

@@ -78,21 +78,19 @@ func (p *Processor) watchEndpoint(svcCtx *servicecontext.Context, id string, ser
 		// We need to inspect the event and get ResourceVersion out of it
 		switch event.Type {
 
-		case watch.Added, watch.Modified:
-			restart, err := epProcessor.AddOrModify(svcCtx, event, &lastKnownGoodEndpoint, service, id,
+		case watch.Added, watch.Modified, watch.Deleted:
+			if event.Type == watch.Deleted {
+				log.Info("[endpoint watcher] endpoint object deleted", "provider", provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace)
+			}
+
+			restart, err := epProcessor.Reconcile(svcCtx, event, &lastKnownGoodEndpoint, service, id,
 				p.StartServicesLeaderElection, &wg, p.clientSet, p.updateEgressConfiguration)
 			if restart {
 				continue
 			} else if err != nil {
-				return fmt.Errorf("[%s] error while processing add/modify event: %w", provider.GetLabel(), err)
+				return fmt.Errorf("[%s] error while processing %s event: %w", provider.GetLabel(), event.Type, err)
 			}
 
-		case watch.Deleted:
-			if err := epProcessor.Delete(svcCtx.Ctx, service, id); err != nil {
-				return fmt.Errorf("[%s] error while processing delete event: %w", provider.GetLabel(), err)
-			}
-
-			log.Info("[endpoint watcher] endpoint object deleted", "provider", provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace)
 		case watch.Error:
 			if svcCtx.Ctx.Err() != nil {
 				return nil