fix: dhcp panic and leak potential

Signed-off-by: Marcel Fest <marcel.fest@telekom.de>
fix(vip): synchronize shared datapath state
2026-09-20 08:03:47 +08:00 · 2026-09-18 14:20:22 +02:00 · 2026-09-18 14:20:22 +02:00 · 2026-09-18 14:20:19 +02:00 · 2026-09-16 18:01:07 +02:00
28 changed files with 2092 additions and 279 deletions
--- a/cmd/kube-vip.go
+++ b/cmd/kube-vip.go
@@ -4,17 +4,14 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"net/http"
 	"os"
 	"slices"
 	"strconv"
 	"strings"
 	"sync"
-	"time"

 	log "log/slog"

-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
@@ -350,12 +347,17 @@ var kubeVipManager = &cobra.Command{
 		ctx, cancel := context.WithCancel(cmd.Context())
 		defer cancel()

+		metrics.RegisterPrometheusMetrics()
+
 		// start prometheus server
 		if initConfig.PrometheusHTTPServer != "" {
 			wg.Go(func() {
-				servePrometheusHTTPServer(ctx, PrometheusHTTPServerConfig{
+				if err := metrics.Serve(ctx, metrics.ServerConfig{
 					Addr: initConfig.PrometheusHTTPServer,
-				})
+				}); err != nil {
+					// Continue even if metrics server fails
+					log.Error("prometheus HTTP server", "err", err)
+				}
 			})
 		}

@@ -473,7 +475,8 @@ var kubeVipManager = &cobra.Command{
 			return fmt.Errorf("new manager: %w", err)
 		}

-		metrics.RegisterPrometheusMetrics()
+		// Label metrics after the call to manager.New, as it may modify the node name
+		// if it was not set in the configuration.
 		metrics.BuildInfo.WithLabelValues(Release.Version, Release.Build, initConfig.NodeName)

 		// Start the service manager, this will watch the config Map and construct kube-vip services for it
@@ -485,65 +488,6 @@ var kubeVipManager = &cobra.Command{
 	},
 }

-// PrometheusHTTPServerConfig defines the Prometheus server configuration.
-type PrometheusHTTPServerConfig struct {
-	// Addr sets the http server address used to expose the metric endpoint
-	Addr string
-}
-
-func servePrometheusHTTPServer(ctx context.Context, config PrometheusHTTPServerConfig) {
-	var err error
-	mux := http.NewServeMux()
-	mux.Handle("/metrics", promhttp.Handler())
-	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { //nolint TODO
-		_, _ = w.Write([]byte(`<html>
-			<head><title>kube-vip</title></head>
-			<body>
-			<h1>kube-vip Metrics</h1>
-			<p><a href="` + "/metrics" + `">Metrics</a></p>
-			</body>
-			</html>`))
-	})
-
-	srv := &http.Server{
-		Addr:              config.Addr,
-		Handler:           mux,
-		ReadHeaderTimeout: 2 * time.Second,
-	}
-
-	wg := sync.WaitGroup{}
-
-	wg.Go(func() {
-		if err = srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-			log.Error("prometheus HTTP server", "err", err)
-			return
-		}
-	})
-
-	log.Info("prometheus HTTP server started")
-
-	<-ctx.Done()
-
-	// create prometheus shutdown context (independent of other contexts)
-	ctxShutDown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer func() {
-		cancel()
-	}()
-
-	if err = srv.Shutdown(ctxShutDown); err != nil {
-		log.Error("shutting down prometheus HTTP server", "err", err)
-		return
-	}
-
-	if err == http.ErrServerClosed {
-		err = nil
-	}
-
-	log.Info("prometheus HTTP server stopped")
-
-	wg.Wait()
-}
-
 func GenerateCidrRange(address string, dnsMode string) (string, error) {
 	var cidrs []string

--- a/go.mod
+++ b/go.mod
@@ -22,7 +22,8 @@ require (
 	github.com/prometheus/client_golang v1.24.1
 	github.com/spf13/cobra v1.10.2
 	github.com/stretchr/testify v1.12.1
-	github.com/vishvananda/netlink v1.3.1
+	github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b
+	github.com/vishvananda/netns v0.0.5
 	go.etcd.io/etcd/api/v3 v3.7.1
 	go.etcd.io/etcd/client/pkg/v3 v3.7.1
 	go.etcd.io/etcd/client/v3 v3.7.1
@@ -120,7 +121,6 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
-	github.com/vishvananda/netns v0.0.5 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
--- a/go.sum
+++ b/go.sum
@@ -265,8 +265,8 @@ github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701 h1:pyC9PaHYZFgEKFdlp3G8
 github.com/u-root/uio v0.0.0-20240224005618-d2acac8f3701/go.mod h1:P3a5rG4X7tI17Nn3aOIAYr5HbIMukwXG0urG0WuL8OA=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
-github.com/vishvananda/netlink v1.3.1 h1:3AEMt62VKqz90r0tmNhog0r/PpWKmrEShJU0wJW6bV0=
-github.com/vishvananda/netlink v1.3.1/go.mod h1:ARtKouGSTGchR8aMwmkzC0qiNPrrWO5JS/XMVl45+b4=
+github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b h1:XtEhFJO3IqjQWHJZ3bbNm7LtbDehriJK65KW+6lnw+Q=
+github.com/vishvananda/netlink v1.3.2-0.20260830232854-cf01b55a4a4b/go.mod h1:lEui7SPMd9fgxzHVGRAvTxsBGCF6PRH81o2kLWLWHgw=
 github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zdEY=
 github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
@@ -331,9 +331,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
 golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
--- a/pkg/election/election.go
+++ b/pkg/election/election.go
@@ -59,7 +59,7 @@ func NewManager(config *kubevip.Config, k8sClientset, rwClientset *kubernetes.Cl
 func RunOrDie(ctx context.Context, run *RunConfig, c *kubevip.Config) error {
 	switch c.LeaderElectionType {
 	case "kubernetes", "":
-		runKubernetesLeaderElectionOrDie(ctx, run)
+		return runKubernetesLeaderElectionOrDie(ctx, run)
 	case "etcd":
 		if err := runEtcdLeaderElectionOrDie(ctx, run); err != nil {
 			return err
@@ -71,20 +71,25 @@ func RunOrDie(ctx context.Context, run *RunConfig, c *kubevip.Config) error {
 	return nil
 }

-func runKubernetesLeaderElectionOrDie(ctx context.Context, run *RunConfig) {
+func runKubernetesLeaderElectionOrDie(ctx context.Context, run *RunConfig) error {
+	annotations, err := kubevip.WithLeaseVIPs(run.LeaseAnnotations, run.Config.InstanceName, run.Config.RoutingProtocol, run.VIPs)
+	if err != nil {
+		return err
+	}
+	leaseClient := run.Mgr.KubernetesClient.CoordinationV1().Leases(run.LeaseID.Namespace())
 	// we use the Lease lock type since edits to Leases are less common
 	// and fewer objects in the cluster watch "all Leases".
-	lock := &resourcelock.LeaseLock{
+	baseLock := &resourcelock.LeaseLock{
 		LeaseMeta: metav1.ObjectMeta{
-			Name:        run.LeaseID.Name(),
-			Namespace:   run.LeaseID.Namespace(),
-			Annotations: run.LeaseAnnotations,
+			Name:      run.LeaseID.Name(),
+			Namespace: run.LeaseID.Namespace(),
 		},
 		Client: run.Mgr.KubernetesClient.CoordinationV1(),
 		LockConfig: resourcelock.ResourceLockConfig{
 			Identity: run.Config.NodeName,
 		},
 	}
+	lock := newAnnotatedLeaseLock(baseLock, leaseClient, run.LeaseID.Name(), annotations)

 	// start the leader election code loop
 	leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
@@ -105,6 +110,7 @@ func runKubernetesLeaderElectionOrDie(ctx context.Context, run *RunConfig) {
 			OnNewLeader:      run.OnNewLeader,
 		},
 	})
+	return nil
 }

 func runEtcdLeaderElectionOrDie(ctx context.Context, run *RunConfig) error {
@@ -135,6 +141,7 @@ type RunConfig struct {
 	LeaseID          lease.ID
 	Mgr              *Manager
 	LeaseAnnotations map[string]string
+	VIPs             []string

 	// onStartedLeading is called when this member starts leading.
 	OnStartedLeading func(context.Context)
--- a/pkg/election/lease_lock.go
+++ b/pkg/election/lease_lock.go
@@ -0,0 +1,92 @@
+package election
+
+import (
+	"context"
+
+	log "log/slog"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	coordinationv1client "k8s.io/client-go/kubernetes/typed/coordination/v1"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/util/retry"
+)
+
+type annotatedLeaseLock struct {
+	resourcelock.Interface
+	leases      coordinationv1client.LeaseInterface
+	name        string
+	annotations map[string]string
+}
+
+func newAnnotatedLeaseLock(lock resourcelock.Interface, leases coordinationv1client.LeaseInterface,
+	name string, annotations map[string]string) resourcelock.Interface {
+	return &annotatedLeaseLock{Interface: lock, leases: leases, name: name, annotations: annotations}
+}
+
+func (lock *annotatedLeaseLock) Create(ctx context.Context, record resourcelock.LeaderElectionRecord) error {
+	if err := lock.Interface.Create(ctx, record); err != nil {
+		return err
+	}
+	lock.ensure(ctx, record)
+	return nil
+}
+
+func (lock *annotatedLeaseLock) Update(ctx context.Context, record resourcelock.LeaderElectionRecord) error {
+	if err := lock.Interface.Update(ctx, record); err != nil {
+		return err
+	}
+	lock.ensure(ctx, record)
+	return nil
+}
+
+// ensure applies the configured annotations once this process holds the lease. Failures are
+// logged rather than returned: the lease write already succeeded, so reporting an error would
+// make the elector stand down while it still holds the lease.
+func (lock *annotatedLeaseLock) ensure(ctx context.Context, record resourcelock.LeaderElectionRecord) {
+	if record.HolderIdentity != lock.Identity() {
+		return
+	}
+	changed, err := lock.ensureAnnotations(ctx)
+	if err != nil {
+		log.Warn("failed to annotate lease", "lease", lock.name, "err", err)
+		return
+	}
+	if !changed {
+		return
+	}
+	// Annotating out of band bumps the resourceVersion, so refresh the wrapped lock's
+	// cached lease or its next optimistic Update conflicts.
+	if _, _, err := lock.Interface.Get(ctx); err != nil {
+		log.Warn("failed to refresh lease after annotating", "lease", lock.name, "err", err)
+	}
+}
+
+func (lock *annotatedLeaseLock) ensureAnnotations(ctx context.Context) (bool, error) {
+	changed := false
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		resource, err := lock.leases.Get(ctx, lock.name, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		if resource.Annotations == nil {
+			resource.Annotations = make(map[string]string, len(lock.annotations))
+		}
+		resourceChanged := false
+		for key, value := range lock.annotations {
+			if resource.Annotations[key] == value {
+				continue
+			}
+			resource.Annotations[key] = value
+			resourceChanged = true
+		}
+		if !resourceChanged {
+			return nil
+		}
+		_, err = lock.leases.Update(ctx, resource, metav1.UpdateOptions{})
+		if err == nil {
+			changed = true
+		}
+		return err
+	})
+	return changed, err
+}
--- a/pkg/election/lease_lock_test.go
+++ b/pkg/election/lease_lock_test.go
@@ -0,0 +1,177 @@
+package election
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/kube-vip/kube-vip/pkg/kubevip"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+	k8stesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+func TestAnnotatedLeaseLockPersistsAnnotationsOnCreateAndUpdate(t *testing.T) {
+	client := fake.NewSimpleClientset()
+	leaseClient := client.CoordinationV1().Leases("default")
+	base := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{Name: "lease", Namespace: "default"},
+		Client:    client.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: "node-a",
+		},
+	}
+	annotations, err := kubevip.WithLeaseVIPs(map[string]string{"example.test/preserved": "true"},
+		"release_a", 248, []string{"192.0.2.10"})
+	if err != nil {
+		t.Fatalf("WithLeaseVIPs() error = %v", err)
+	}
+	lock := newAnnotatedLeaseLock(base, leaseClient, "lease", annotations)
+	record := resourcelock.LeaderElectionRecord{HolderIdentity: "node-a"}
+	if err := lock.Create(context.Background(), record); err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+	if err := lock.Update(context.Background(), record); err != nil {
+		t.Fatalf("Update() error = %v", err)
+	}
+
+	resource, err := leaseClient.Get(context.Background(), "lease", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("get Lease: %v", err)
+	}
+	value, err := kubevip.ParseLeaseVIPs(resource.Annotations[kubevip.LeaseVIPs])
+	if err != nil {
+		t.Fatalf("ParseLeaseVIPs() error = %v", err)
+	}
+	if value.InstanceName != "release_a" || value.IFAProto != 248 || len(value.VIPs) != 1 ||
+		value.VIPs[0] != (kubevip.LeaseVIP{Index: 0, Value: "192.0.2.10", Kind: kubevip.LeaseVIPKindAddress}) {
+		t.Fatalf("Lease VIP metadata = %+v", value)
+	}
+	if resource.Annotations["example.test/preserved"] != "true" {
+		t.Fatal("Lease update dropped a configured annotation")
+	}
+}
+
+// A failed annotation write must not be reported to the leader elector: the lease itself
+// was already written, and an error makes the elector stand down while it still holds it.
+func TestAnnotatedLeaseLockAnnotationFailureDoesNotSurfaceToElector(t *testing.T) {
+	client := fake.NewSimpleClientset()
+	base := &resourcelock.LeaseLock{
+		LeaseMeta:  metav1.ObjectMeta{Name: "lease", Namespace: "default"},
+		Client:     client.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{Identity: "node-a"},
+	}
+
+	failing := fake.NewSimpleClientset()
+	failing.PrependReactor("get", "leases", func(k8stesting.Action) (bool, runtime.Object, error) {
+		return true, nil, fmt.Errorf("annotation backend unavailable")
+	})
+
+	annotations, err := kubevip.WithLeaseVIPs(nil, "release_a", 248, []string{"192.0.2.10"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	lock := newAnnotatedLeaseLock(base, failing.CoordinationV1().Leases("default"), "lease", annotations)
+	record := resourcelock.LeaderElectionRecord{HolderIdentity: "node-a"}
+
+	if err := lock.Create(context.Background(), record); err != nil {
+		t.Fatalf("Create() error = %v, want nil so the elector keeps the lease", err)
+	}
+	if err := lock.Update(context.Background(), record); err != nil {
+		t.Fatalf("Update() error = %v, want nil so the elector keeps the lease", err)
+	}
+
+	if _, err := client.CoordinationV1().Leases("default").Get(context.Background(), "lease",
+		metav1.GetOptions{}); err != nil {
+		t.Fatalf("wrapped lock did not write the lease: %v", err)
+	}
+}
+
+func TestAnnotatedLeaseLockFollowerDoesNotOverwriteAnnotations(t *testing.T) {
+	client := fake.NewSimpleClientset()
+	leaseClient := client.CoordinationV1().Leases("default")
+	newBase := func(identity string) *resourcelock.LeaseLock {
+		return &resourcelock.LeaseLock{
+			LeaseMeta:  metav1.ObjectMeta{Name: "lease", Namespace: "default"},
+			Client:     client.CoordinationV1(),
+			LockConfig: resourcelock.ResourceLockConfig{Identity: identity},
+		}
+	}
+	ownerBase := newBase("node-a")
+	followerBase := newBase("node-b")
+	active, err := kubevip.WithLeaseVIPs(nil, "release_a", 248, []string{"192.0.2.10"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	creator := newAnnotatedLeaseLock(ownerBase, leaseClient, "lease", active)
+	if err := creator.Create(context.Background(), resourcelock.LeaderElectionRecord{HolderIdentity: "node-a"}); err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+
+	follower, err := kubevip.WithLeaseVIPs(nil, "release_b", 249, []string{"192.0.2.20"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	observer := newAnnotatedLeaseLock(followerBase, leaseClient, "lease", follower)
+	if _, _, err := observer.Get(context.Background()); err != nil {
+		t.Fatalf("Get() error = %v", err)
+	}
+	resource, err := leaseClient.Get(context.Background(), "lease", metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	metadata, err := kubevip.ParseLeaseVIPs(resource.Annotations[kubevip.LeaseVIPs])
+	if err != nil {
+		t.Fatal(err)
+	}
+	if metadata.InstanceName != "release_a" || metadata.IFAProto != 248 {
+		t.Fatalf("follower overwrote active metadata: %+v", metadata)
+	}
+}
+
+func TestAnnotatedLeaseLockReleaseDoesNotOverwriteSuccessorMetadata(t *testing.T) {
+	client := fake.NewSimpleClientset()
+	leaseClient := client.CoordinationV1().Leases("default")
+	newLock := func(identity, instanceName string, protocol int, vip string) resourcelock.Interface {
+		base := &resourcelock.LeaseLock{
+			LeaseMeta:  metav1.ObjectMeta{Name: "lease", Namespace: "default"},
+			Client:     client.CoordinationV1(),
+			LockConfig: resourcelock.ResourceLockConfig{Identity: identity},
+		}
+		annotations, err := kubevip.WithLeaseVIPs(nil, instanceName, protocol, []string{vip})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return newAnnotatedLeaseLock(base, leaseClient, "lease", annotations)
+	}
+
+	first := newLock("node-a", "release_a", 248, "192.0.2.10")
+	if err := first.Create(context.Background(), resourcelock.LeaderElectionRecord{HolderIdentity: "node-a"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := first.Update(context.Background(), resourcelock.LeaderElectionRecord{}); err != nil {
+		t.Fatal(err)
+	}
+
+	second := newLock("node-b", "release_b", 249, "192.0.2.20")
+	if _, _, err := second.Get(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if err := second.Update(context.Background(), resourcelock.LeaderElectionRecord{HolderIdentity: "node-b"}); err != nil {
+		t.Fatal(err)
+	}
+
+	resource, err := leaseClient.Get(context.Background(), "lease", metav1.GetOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	metadata, err := kubevip.ParseLeaseVIPs(resource.Annotations[kubevip.LeaseVIPs])
+	if err != nil {
+		t.Fatal(err)
+	}
+	if metadata.InstanceName != "release_b" || metadata.IFAProto != 249 || metadata.VIPs[0].Value != "192.0.2.20" {
+		t.Fatalf("successor metadata = %+v", metadata)
+	}
+}
--- a/pkg/kubevip/annotations.go
+++ b/pkg/kubevip/annotations.go
@@ -77,6 +77,9 @@ const (
 	// Name of the service lease object
 	ServiceLease = "kube-vip.io/leaseName"

+	// Versioned kube-vip ownership metadata stored on Kubernetes election Leases
+	LeaseVIPs = "kube-vip.io/lease-vips"
+
 	// Forces kube-vip to use per service election for this particular service
 	ForcePerServiceElection = "kube-vip.io/forcePerServiceElection"

--- a/pkg/kubevip/lease_annotations.go
+++ b/pkg/kubevip/lease_annotations.go
@@ -0,0 +1,134 @@
+package kubevip
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/netip"
+	"slices"
+	"strings"
+)
+
+const LeaseVIPsVersion = "v1"
+
+// LeaseVIPKind distinguishes literal addresses from names that resolve to one.
+type LeaseVIPKind string
+
+const (
+	LeaseVIPKindAddress LeaseVIPKind = "address"
+	LeaseVIPKindName    LeaseVIPKind = "name"
+)
+
+type LeaseVIPsValue struct {
+	Version      string     `json:"version"`
+	InstanceName string     `json:"instance_name"`
+	IFAProto     int        `json:"ifa_proto"`
+	VIPs         []LeaseVIP `json:"vips"`
+}
+
+type LeaseVIP struct {
+	Index int          `json:"index"`
+	Value string       `json:"value"`
+	Kind  LeaseVIPKind `json:"kind"`
+}
+
+func WithLeaseVIPs(annotations map[string]string, instanceName string, ifaProto int, vips []string) (map[string]string, error) {
+	result := make(map[string]string, len(annotations)+1)
+	for key, value := range annotations {
+		result[key] = value
+	}
+
+	encoded, err := json.Marshal(LeaseVIPsValue{
+		Version:      LeaseVIPsVersion,
+		InstanceName: instanceName,
+		IFAProto:     ifaProto,
+		VIPs:         normalizeLeaseVIPs(vips),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("encode %s annotation: %w", LeaseVIPs, err)
+	}
+	result[LeaseVIPs] = string(encoded)
+	return result, nil
+}
+
+func ParseLeaseVIPs(value string) (LeaseVIPsValue, error) {
+	var parsed LeaseVIPsValue
+	if err := json.Unmarshal([]byte(value), &parsed); err != nil {
+		return LeaseVIPsValue{}, fmt.Errorf("decode %s annotation: %w", LeaseVIPs, err)
+	}
+	if parsed.Version != LeaseVIPsVersion {
+		return LeaseVIPsValue{}, fmt.Errorf("unsupported %s annotation version %q", LeaseVIPs, parsed.Version)
+	}
+	for index, vip := range parsed.VIPs {
+		if vip.Index != index {
+			return LeaseVIPsValue{}, fmt.Errorf("invalid %s VIP index %d at position %d", LeaseVIPs, vip.Index, index)
+		}
+		switch vip.Kind {
+		case LeaseVIPKindAddress, LeaseVIPKindName:
+		default:
+			return LeaseVIPsValue{}, fmt.Errorf("invalid %s VIP kind %q at index %d", LeaseVIPs, vip.Kind, vip.Index)
+		}
+	}
+	return parsed, nil
+}
+
+func normalizeLeaseVIPs(values []string) []LeaseVIP {
+	unique := make(map[string]struct{}, len(values))
+	addresses := make([]string, 0, len(values))
+	for _, value := range values {
+		for candidate := range strings.SplitSeq(value, ",") {
+			candidate = strings.TrimSpace(candidate)
+			if candidate == "" {
+				continue
+			}
+			if _, exists := unique[candidate]; exists {
+				continue
+			}
+			unique[candidate] = struct{}{}
+			addresses = append(addresses, candidate)
+		}
+	}
+	// Sorting keeps the annotation byte-identical however callers happen to order VIPs.
+	slices.SortFunc(addresses, compareLeaseVIPs)
+
+	result := make([]LeaseVIP, 0, len(addresses))
+	for _, address := range addresses {
+		kind := LeaseVIPKindName
+		if _, isAddress := leaseVIPAddress(address); isAddress {
+			kind = LeaseVIPKindAddress
+		}
+		result = append(result, LeaseVIP{Index: len(result), Value: address, Kind: kind})
+	}
+	return result
+}
+
+// compareLeaseVIPs orders addresses numerically and ahead of names, which keeps VIPs like
+// 10.0.0.2 and 10.0.0.10 in the order an operator expects. Values that are not addresses,
+// such as DNS records, are kept and ordered lexically.
+func compareLeaseVIPs(a, b string) int {
+	addressA, isAddressA := leaseVIPAddress(a)
+	addressB, isAddressB := leaseVIPAddress(b)
+	switch {
+	case isAddressA && isAddressB:
+		if order := addressA.Compare(addressB); order != 0 {
+			return order
+		}
+		// Distinct spellings of one address still need a stable order.
+		return strings.Compare(a, b)
+	case isAddressA:
+		return -1
+	case isAddressB:
+		return 1
+	default:
+		return strings.Compare(a, b)
+	}
+}
+
+func leaseVIPAddress(value string) (netip.Addr, bool) {
+	if address, err := netip.ParseAddr(value); err == nil {
+		return address.Unmap(), true
+	}
+	if prefix, err := netip.ParsePrefix(value); err == nil {
+		return prefix.Addr().Unmap(), true
+	}
+	return netip.Addr{}, false
+}
--- a/pkg/kubevip/lease_annotations_test.go
+++ b/pkg/kubevip/lease_annotations_test.go
@@ -0,0 +1,94 @@
+package kubevip
+
+import (
+	"slices"
+	"testing"
+)
+
+func TestWithLeaseVIPsEncodesVersionedInstanceOwnership(t *testing.T) {
+	base := map[string]string{"example.test/preserved": "true", LeaseVIPs: "stale"}
+	annotations, err := WithLeaseVIPs(base, "release_a", 248, []string{
+		"2001:db8::10/128", "192.0.2.10", "192.0.2.10/32", "api.example.test",
+	})
+	if err != nil {
+		t.Fatalf("WithLeaseVIPs() error = %v", err)
+	}
+	if annotations["example.test/preserved"] != "true" {
+		t.Fatal("WithLeaseVIPs() dropped an existing annotation")
+	}
+	if base[LeaseVIPs] != "stale" {
+		t.Fatal("WithLeaseVIPs() mutated the input annotations")
+	}
+
+	value, err := ParseLeaseVIPs(annotations[LeaseVIPs])
+	if err != nil {
+		t.Fatalf("ParseLeaseVIPs() error = %v", err)
+	}
+	if value.Version != LeaseVIPsVersion || value.InstanceName != "release_a" || value.IFAProto != 248 {
+		t.Fatalf("Lease VIP metadata = %+v", value)
+	}
+	// Values are stored verbatim so DNS records survive alongside addresses.
+	want := []LeaseVIP{
+		{Index: 0, Value: "192.0.2.10", Kind: LeaseVIPKindAddress},
+		{Index: 1, Value: "192.0.2.10/32", Kind: LeaseVIPKindAddress},
+		{Index: 2, Value: "2001:db8::10/128", Kind: LeaseVIPKindAddress},
+		{Index: 3, Value: "api.example.test", Kind: LeaseVIPKindName},
+	}
+	if !slices.Equal(value.VIPs, want) {
+		t.Fatalf("Lease VIPs = %v, want %v", value.VIPs, want)
+	}
+}
+
+// The annotation is rewritten whenever a node starts campaigning, so the encoding
+// has to be stable even when callers collect the same VIPs in a different order.
+func TestWithLeaseVIPsIsIndependentOfInputOrder(t *testing.T) {
+	first, err := WithLeaseVIPs(nil, "release_a", 248, []string{
+		"2001:db8::10", "192.0.2.10", "10.0.0.2", "10.0.0.10",
+	})
+	if err != nil {
+		t.Fatalf("WithLeaseVIPs() error = %v", err)
+	}
+	second, err := WithLeaseVIPs(nil, "release_a", 248, []string{
+		"10.0.0.10", "192.0.2.10", "2001:db8::10", "10.0.0.2",
+	})
+	if err != nil {
+		t.Fatalf("WithLeaseVIPs() error = %v", err)
+	}
+	if first[LeaseVIPs] != second[LeaseVIPs] {
+		t.Fatalf("annotation changed with input order:\n%s\n%s", first[LeaseVIPs], second[LeaseVIPs])
+	}
+
+	value, err := ParseLeaseVIPs(first[LeaseVIPs])
+	if err != nil {
+		t.Fatalf("ParseLeaseVIPs() error = %v", err)
+	}
+	want := []string{"10.0.0.2", "10.0.0.10", "192.0.2.10", "2001:db8::10"}
+	if len(value.VIPs) != len(want) {
+		t.Fatalf("Lease VIPs = %v, want %v", value.VIPs, want)
+	}
+	for index, address := range want {
+		if value.VIPs[index] != (LeaseVIP{Index: index, Value: address, Kind: LeaseVIPKindAddress}) {
+			t.Fatalf("Lease VIPs = %v, want %v", value.VIPs, want)
+		}
+	}
+}
+
+func TestParseLeaseVIPsRejectsUnknownVersion(t *testing.T) {
+	if _, err := ParseLeaseVIPs(`{"version":"v2","instance_name":"release_a","ifa_proto":248,"vips":[]}`); err == nil {
+		t.Fatal("ParseLeaseVIPs() accepted an unknown version")
+	}
+}
+
+func TestParseLeaseVIPsRejectsUnknownKind(t *testing.T) {
+	if _, err := ParseLeaseVIPs(
+		`{"version":"v1","instance_name":"release_a","ifa_proto":248,"vips":[{"index":0,"value":"192.0.2.10","kind":"cidr"}]}`,
+	); err == nil {
+		t.Fatal("ParseLeaseVIPs() accepted an unknown VIP kind")
+	}
+}
+
+func TestParseLeaseVIPsRejectsOutOfOrderIndexes(t *testing.T) {
+	if _, err := ParseLeaseVIPs(`{"version":"v1","instance_name":"release_a","ifa_proto":248,"vips":[{"index":1,"value":"192.0.2.10"}]}`); err == nil {
+		t.Fatal("ParseLeaseVIPs() accepted an out-of-order VIP index")
+	}
+}
--- a/pkg/manager/manager.go
+++ b/pkg/manager/manager.go
@@ -32,7 +32,6 @@ import (
 	"github.com/kube-vip/kube-vip/pkg/upnp"
 	"github.com/kube-vip/kube-vip/pkg/utils"
 	"github.com/kube-vip/kube-vip/pkg/vip"
-	"github.com/prometheus/client_golang/prometheus"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 )
@@ -57,10 +56,6 @@ type Manager struct {

 	svcProcessor *services.Processor

-	// This is a prometheus counter used to count the number of events received
-	// from the service watcher
-	countServiceWatchEvent *prometheus.CounterVec
-
 	// This mutex is to protect calls from various goroutines
 	mutex sync.Mutex

@@ -259,16 +254,10 @@ func New(ctx context.Context, configMap string, config *kubevip.Config) (*Manage
 		intfMgr, arpMgr, nodeLabelManager, electionMgr, leaseMgr, routeMgr)

 	return &Manager{
-		clientSet:   clientset,
-		rwClientSet: rwClientSet,
-		configMap:   configMap,
-		config:      config,
-		countServiceWatchEvent: prometheus.NewCounterVec(prometheus.CounterOpts{
-			Namespace: "kube_vip",
-			Subsystem: "manager",
-			Name:      "all_services_events",
-			Help:      "Count all events fired by the service watcher categorised by event type",
-		}, []string{"type"}),
+		clientSet:        clientset,
+		rwClientSet:      rwClientSet,
+		configMap:        configMap,
+		config:           config,
 		signalChan:       signalChan,
 		svcProcessor:     svcProcessor,
 		intfMgr:          intfMgr,
--- a/pkg/metrics/prometheus.go
+++ b/pkg/metrics/prometheus.go
@@ -1,6 +1,10 @@
 package metrics

-import "github.com/prometheus/client_golang/prometheus"
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)

 var (
 	// Service / VIP Lifecycle
@@ -68,19 +72,25 @@ var (
 	)
 )

+var registerOnce sync.Once
+
+// RegisterPrometheusMetrics registers all kube-vip metrics with the default
+// Prometheus registry.
 func RegisterPrometheusMetrics() {
-	// Register all metrics with Prometheus
-	prometheus.MustRegister(
-		ActiveServices,
-		ServiceReconcileErrorsTotal,
-		ServiceReconcileDuration,
-		LeaderTransitionsTotal,
-		IsLeader,
-		ServiceElectionLoops,
-		ServiceElectionAttemptsTotal,
-		ServiceElectionErrorsTotal,
-		BGPSessionInfoGauge,
-		BuildInfo,
-		CountServiceWatchEvent,
-	)
+	registerOnce.Do(func() {
+		// Register all metrics with Prometheus
+		prometheus.MustRegister(
+			ActiveServices,
+			ServiceReconcileErrorsTotal,
+			ServiceReconcileDuration,
+			LeaderTransitionsTotal,
+			IsLeader,
+			ServiceElectionLoops,
+			ServiceElectionAttemptsTotal,
+			ServiceElectionErrorsTotal,
+			BGPSessionInfoGauge,
+			BuildInfo,
+			CountServiceWatchEvent,
+		)
+	})
 }
--- a/pkg/metrics/server.go
+++ b/pkg/metrics/server.go
@@ -0,0 +1,94 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	log "log/slog"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// shutdownTimeout bounds how long the server waits for in-flight requests to
+// finish once the context is cancelled.
+const shutdownTimeout = 5 * time.Second
+
+// ServerConfig defines the Prometheus server configuration.
+type ServerConfig struct {
+	// Addr sets the http server address used to expose the metric endpoint
+	Addr string
+}
+
+// Serve exposes the Prometheus metrics endpoint on the configured address.
+func Serve(ctx context.Context, config ServerConfig) error {
+	ln, err := net.Listen("tcp", config.Addr)
+	if err != nil {
+		return fmt.Errorf("listening on %q: %w", config.Addr, err)
+	}
+
+	return serve(ctx, ln)
+}
+
+// serve starts the metrics endpoint on the provided listener
+func serve(ctx context.Context, ln net.Listener) error {
+	srv := &http.Server{
+		Handler:           newServeMux(),
+		ReadHeaderTimeout: 2 * time.Second,
+	}
+
+	wg := sync.WaitGroup{}
+	defer wg.Wait()
+
+	serveErr := make(chan error, 1)
+	wg.Go(func() {
+		err := srv.Serve(ln)
+		if errors.Is(err, http.ErrServerClosed) {
+			err = nil
+		}
+		serveErr <- err
+	})
+
+	log.Info("prometheus HTTP server started", "addr", ln.Addr().String())
+
+	select {
+	case err := <-serveErr:
+		if err != nil {
+			return fmt.Errorf("serving prometheus metrics: %w", err)
+		}
+		return nil
+	case <-ctx.Done():
+	}
+
+	// create prometheus shutdown context (independent of other contexts)
+	ctxShutDown, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
+	defer cancel()
+
+	if err := srv.Shutdown(ctxShutDown); err != nil {
+		return fmt.Errorf("shutting down prometheus HTTP server: %w", err)
+	}
+
+	log.Info("prometheus HTTP server stopped")
+
+	return nil
+}
+
+func newServeMux() *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+	mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte(`<html>
+			<head><title>kube-vip</title></head>
+			<body>
+			<h1>kube-vip Metrics</h1>
+			<p><a href="/metrics">Metrics</a></p>
+			</body>
+			</html>`))
+	})
+
+	return mux
+}
--- a/pkg/metrics/server_test.go
+++ b/pkg/metrics/server_test.go
@@ -0,0 +1,213 @@
+package metrics
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+func TestServeExposesKubeVipMetrics(t *testing.T) {
+	RegisterPrometheusMetrics()
+	version, build, node := "v1.2.3", "test-build", "node-1"
+	BuildInfo.WithLabelValues(version, build, node)
+
+	base, stop := startServer(t, newTestListener(t))
+
+	body, code := get(t, base+"/metrics")
+	if code != http.StatusOK {
+		t.Fatalf("GET /metrics status = %d, want %d", code, http.StatusOK)
+	}
+
+	// Label names are exposed in alphabetical order.
+	want := fmt.Sprintf("kube_vip_build_info{build=\"%s\",node=\"%s\",version=\"%s\"}", build, node, version)
+	if !strings.Contains(body, want) {
+		t.Errorf("GET /metrics body does not contain %s, got:\n%s", want, body)
+	}
+
+	if err := stop(); err != nil {
+		t.Errorf("serve returned an error on shutdown: %v", err)
+	}
+}
+
+func TestServeRootPageLinksToMetrics(t *testing.T) {
+	base, stop := startServer(t, newTestListener(t))
+
+	body, code := get(t, base+"/")
+	if code != http.StatusOK {
+		t.Fatalf("GET / status = %d, want %d", code, http.StatusOK)
+	}
+
+	if !strings.Contains(body, `href="/metrics"`) {
+		t.Errorf("GET / body does not link to /metrics, got:\n%s", body)
+	}
+
+	if err := stop(); err != nil {
+		t.Errorf("serve returned an error on shutdown: %v", err)
+	}
+}
+
+func TestServeStopsOnContextCancellation(t *testing.T) {
+	ln := newTestListener(t)
+	addr := ln.Addr().String()
+
+	_, stop := startServer(t, ln)
+
+	// stop blocks until serve returns, and serve waits on its serving
+	// goroutine, so a clean return means nothing was left running.
+	if err := stop(); err != nil {
+		t.Fatalf("serve returned an error on shutdown: %v", err)
+	}
+
+	// Shutdown must have closed the listener, freeing the port.
+	reopened, err := net.Listen("tcp", addr)
+	if err != nil {
+		t.Fatalf("listener still bound to %s after shutdown: %v", addr, err)
+	}
+	_ = reopened.Close()
+}
+
+func TestServeWithAlreadyCancelledContext(t *testing.T) {
+	ln := newTestListener(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	// Shutdown can win the race against the serving goroutine here. That is
+	// safe: a server already told to shut down makes Serve return
+	// ErrServerClosed straight away, so nothing blocks.
+	done := make(chan error, 1)
+	go func() {
+		done <- serve(ctx, ln)
+	}()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("serve on an already cancelled context returned: %v", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("serve hung on an already cancelled context")
+	}
+}
+
+func TestServeReturnsErrorWhenAddressUnavailable(t *testing.T) {
+	ln := newTestListener(t)
+	defer ln.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// The port is already used in newTestListener, so Serve should return an error
+	if err := Serve(ctx, ServerConfig{Addr: ln.Addr().String()}); err == nil {
+		t.Fatal("Serve on an address already in use returned no error")
+	}
+}
+
+func TestRegisterPrometheusMetricsIsIdempotent(t *testing.T) {
+	RegisterPrometheusMetrics()
+
+	// Registering a collector that is already registered is an error,
+	// RegisterPrometheusMetrics should be guarded with sync.Once.
+	mustNotPanic(t, "repeated RegisterPrometheusMetrics call", RegisterPrometheusMetrics)
+
+	// Confirm the collectors were really registered.
+	err := prometheus.DefaultRegisterer.Register(ActiveServices)
+
+	var alreadyRegistered prometheus.AlreadyRegisteredError
+	if !errors.As(err, &alreadyRegistered) {
+		t.Fatalf("Register(ActiveServices) error = %v, want AlreadyRegisteredError", err)
+	}
+}
+
+// newTestListener binds a loopback listener on an arbitrary free port.
+func newTestListener(t *testing.T) net.Listener {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listening on a free loopback port: %v", err)
+	}
+
+	return ln
+}
+
+// startServer runs serve on ln and returns the base URL along with a stop
+// function that cancels the context and reports what serve returned.
+func startServer(t *testing.T, ln net.Listener) (string, func() error) {
+	t.Helper()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	serveErr := make(chan error, 1)
+	go func() {
+		serveErr <- serve(ctx, ln)
+	}()
+
+	var (
+		once sync.Once
+		err  error
+	)
+	stop := func() error {
+		once.Do(func() {
+			cancel()
+			select {
+			case err = <-serveErr:
+			case <-time.After(10 * time.Second):
+				err = errors.New("serve did not return after the context was cancelled")
+			}
+		})
+		return err
+	}
+	t.Cleanup(func() {
+		_ = stop()
+	})
+
+	return "http://" + ln.Addr().String(), stop
+}
+
+func get(t *testing.T, url string) (string, int) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		t.Fatalf("building request for %s: %v", url, err)
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("GET %s: %v", url, err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading body of %s: %v", url, err)
+	}
+
+	return string(body), resp.StatusCode
+}
+
+// mustNotPanic reports a panic in fn as a test failure describing what
+// panicked, rather than letting it take down the test binary.
+func mustNotPanic(t *testing.T, what string, fn func()) {
+	t.Helper()
+
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("%s panicked: %v", what, r)
+		}
+	}()
+
+	fn()
+}
--- a/pkg/networkinterface/networkinterface.go
+++ b/pkg/networkinterface/networkinterface.go
@@ -1,19 +1,19 @@
 package networkinterface

 import (
-	log "log/slog"
 	"sync"

 	"github.com/vishvananda/netlink"
 )

 type Manager struct {
+	lock       sync.Mutex
 	interfaces map[string]*Link
 }

 type Link struct {
-	Lock sync.Mutex
-	Intf netlink.Link
+	mu   sync.Mutex
+	intf netlink.Link
 }

 func NewManager() *Manager {
@@ -23,19 +23,37 @@ func NewManager() *Manager {
 }

 func (m *Manager) Get(intf netlink.Link) *Link {
-	if l, ok := m.interfaces[intf.Attrs().Name]; ok {
-		updated, err := netlink.LinkByName(l.Intf.Attrs().Name)
-		if err != nil {
-			log.Error("failed to get interface %q: %w", l.Intf.Attrs().Name, err)
-			return nil
-		}
-		l.Intf = updated
-		return l
+	if intf == nil || intf.Attrs() == nil {
+		return nil
 	}
-	result := &Link{
-		Intf: intf,
+	attrs := intf.Attrs()
+
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if link, ok := m.interfaces[attrs.Name]; ok {
+		link.replace(intf)
+		return link
 	}

-	m.interfaces[intf.Attrs().Name] = result
-	return result
+	link := &Link{intf: intf}
+	m.interfaces[attrs.Name] = link
+	return link
+}
+
+func (l *Link) WithInterface(run func(netlink.Link) error) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return run(l.intf)
+}
+
+func (l *Link) replace(intf netlink.Link) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.intf = intf
+}
+
+func (m *Manager) Len() int {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	return len(m.interfaces)
 }
--- a/pkg/networkinterface/networkinterface_instance_test.go
+++ b/pkg/networkinterface/networkinterface_instance_test.go
@@ -0,0 +1,65 @@
+package networkinterface_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kube-vip/kube-vip/pkg/arp"
+	"github.com/kube-vip/kube-vip/pkg/instance"
+	"github.com/kube-vip/kube-vip/pkg/kubevip"
+	"github.com/kube-vip/kube-vip/pkg/networkinterface"
+	"github.com/kube-vip/kube-vip/pkg/node/noop"
+	"github.com/kube-vip/kube-vip/pkg/route"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestManagerReconstructsProductionInstanceConcurrently(t *testing.T) {
+	config := &kubevip.Config{Interface: "lo", ServicesInterface: "lo", VIPSubnet: "32", DisableServiceUpdates: true}
+	manager := networkinterface.NewManager()
+	service := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: "service", Namespace: "default", UID: "service"},
+		Spec:       v1.ServiceSpec{LoadBalancerIP: "192.0.2.10"},
+	}
+
+	start := make(chan struct{})
+	var ready sync.WaitGroup
+	ready.Add(2)
+	results := make(chan struct {
+		instance *instance.Instance
+		err      error
+	}, 2)
+	for range 2 {
+		go func() {
+			ready.Done()
+			<-start
+			instanceConfig := *config
+			created, err := instance.NewInstance(context.Background(), service.DeepCopy(), &instanceConfig, manager,
+				arp.NewManager(&instanceConfig), route.NewManager(), noop.NewManager(), &sync.WaitGroup{})
+			results <- struct {
+				instance *instance.Instance
+				err      error
+			}{created, err}
+		}()
+	}
+	ready.Wait()
+	close(start)
+	for range 2 {
+		select {
+		case result := <-results:
+			if result.err != nil {
+				t.Fatalf("NewInstance() error = %v", result.err)
+			}
+			if len(result.instance.Clusters) != 1 {
+				t.Fatalf("cluster count = %d, want 1", len(result.instance.Clusters))
+			}
+		case <-time.After(3 * time.Second):
+			t.Fatal("timed out waiting for concurrent NewInstance calls")
+		}
+	}
+	if got := manager.Len(); got != 1 {
+		t.Fatalf("cached link count = %d, want 1", got)
+	}
+}
--- a/pkg/networkinterface/networkinterface_test.go
+++ b/pkg/networkinterface/networkinterface_test.go
@@ -0,0 +1,74 @@
+package networkinterface
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/vishvananda/netlink"
+)
+
+func TestManagerGetReplacesChangedInterfaceIndex(t *testing.T) {
+	manager := NewManager()
+	firstInterface := dummyLink("eth0", 1)
+	first := manager.Get(firstInterface)
+
+	if got := manager.Get(dummyLink("eth0", 1)); got != first {
+		t.Fatal("Get returned a new link for the same interface generation")
+	}
+
+	secondInterface := dummyLink("eth0", 2)
+	second := manager.Get(secondInterface)
+	if second != first {
+		t.Fatal("Get replaced the shared link after the interface index changed")
+	}
+	var current netlink.Link
+	if err := first.WithInterface(func(intf netlink.Link) error {
+		current = intf
+		return nil
+	}); err != nil {
+		t.Fatalf("WithInterface() error = %v", err)
+	}
+	if current != secondInterface {
+		t.Fatal("Get did not retain the new link generation")
+	}
+}
+
+func TestManagerGetConcurrent(t *testing.T) {
+	manager := NewManager()
+	interfaces := []netlink.Link{dummyLink("eth0", 1), dummyLink("eth1", 2)}
+	var wg sync.WaitGroup
+	results := make(chan struct {
+		index int
+		link  *Link
+	}, 64)
+	for index := range cap(results) {
+		interfaceIndex := index % len(interfaces)
+		wg.Go(func() {
+			results <- struct {
+				index int
+				link  *Link
+			}{index: interfaceIndex, link: manager.Get(interfaces[interfaceIndex])}
+		})
+	}
+	wg.Wait()
+	close(results)
+
+	var cached [2]*Link
+	for result := range results {
+		if result.link == nil {
+			t.Fatal("concurrent interface lookup returned nil")
+		}
+		if cached[result.index] == nil {
+			cached[result.index] = result.link
+		} else if result.link != cached[result.index] {
+			t.Fatalf("interface %d produced multiple cached Link objects", result.index)
+		}
+	}
+	if cached[0] == cached[1] {
+		t.Fatal("different interfaces shared one cached Link object")
+	}
+}
+
+func dummyLink(name string, index int) netlink.Link {
+	return &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: name, Index: index}}
+}
--- a/pkg/route/manager.go
+++ b/pkg/route/manager.go
@@ -73,6 +73,7 @@ func (m *Manager) Add(object string, r route, precheck, update bool) error {
 		if added {
 			log.Debug("[RT] added route", "path", key, "object", object)
 		}
+		m.tracker[key] = itm
 	}

 	itm.objects[object] = true
@@ -112,6 +113,8 @@ func (m *Manager) Delete(object string, r route) error {
 }

 func (m *Manager) Clear() {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	for _, itm := range m.tracker {
 		if err := itm.route.DeleteRoute(); err != nil {
 			log.Warn("[RT] failed to delete route", "err", err.Error())
@@ -121,6 +124,8 @@ func (m *Manager) Clear() {
 }

 func (m *Manager) Check(key string) bool {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
 	_, exists := m.tracker[key]
 	return exists
 }
--- a/pkg/route/manager_test.go
+++ b/pkg/route/manager_test.go
@@ -1,7 +1,9 @@
 package route

 import (
+	"errors"
 	"fmt"
+	"sync"
 	"testing"
 )

@@ -184,6 +186,47 @@ func Test_MultipleRoutesAddDel(t *testing.T) {
 	})
 }

+func TestAddFailureDoesNotTrackRoute(t *testing.T) {
+	m := NewManager()
+	r := &mockRoute{hash: "failed-route", addErr: errors.New("add failed")}
+
+	if err := m.Add("service", r, false, false); err == nil {
+		t.Fatal("Add error = nil, want route failure")
+	}
+	if m.Check(r.RouteHash()) {
+		t.Fatal("failed route was tracked")
+	}
+
+	r.addErr = nil
+	if err := m.Add("service", r, false, false); err != nil {
+		t.Fatalf("retry Add error = %v", err)
+	}
+	if !m.Check(r.RouteHash()) {
+		t.Fatal("successful retry was not tracked")
+	}
+}
+
+func TestClearAndCheckAreSafeWithRouteUpdates(t *testing.T) {
+	manager := NewManager()
+	route := &mockRoute{hash: "concurrent-route", added: true}
+	if err := manager.Add("service", route, false, false); err != nil {
+		t.Fatalf("Add() error = %v", err)
+	}
+
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Go(func() {
+			manager.Check(route.RouteHash())
+		})
+	}
+	wg.Go(manager.Clear)
+	wg.Wait()
+
+	if manager.Check(route.RouteHash()) {
+		t.Fatal("route remained tracked after concurrent Clear")
+	}
+}
+
 type mockRoute struct {
 	added     bool
 	addCalls  int
--- a/pkg/services/services.go
+++ b/pkg/services/services.go
@@ -227,17 +227,22 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
 			if index == -1 {
 				log.Error("unable to find proper VIPConfig for the DHCPv4")
 			} else {
-				for ip := range inst.DHCPv4Client.IPChannel() {
-					log.Debug("IP changed", "ip", ip)
-					inst.VIPConfigs[index].VIP = ip
-					inst.DHCPInterfaceIPv4 = ip
-					if !p.config.DisableServiceUpdates {
-						if err := p.updateStatus(ctx, inst); err != nil {
-							log.Warn("updating svc", "err", err)
+				for {
+					select {
+					case <-ctx.Done():
+						log.Debug("IPv4 update watcher stopping")
+						return
+					case ip := <-inst.DHCPv4Client.IPChannel():
+						log.Debug("IP changed", "ip", ip)
+						inst.VIPConfigs[index].VIP = ip
+						inst.DHCPInterfaceIPv4 = ip
+						if !p.config.DisableServiceUpdates {
+							if err := p.updateStatus(ctx, inst); err != nil {
+								log.Warn("updating svc", "err", err)
+							}
 						}
 					}
 				}
-				log.Debug("IPv4 update channel closed, stopping")
 			}
 		})
 	}
@@ -255,17 +260,22 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc
 			if index == -1 {
 				log.Error("unable to find proper VIPConfig for the DHCPv6")
 			} else {
-				for ip := range inst.DHCPv6Client.IPChannel() {
-					log.Debug("IP changed", "ip", ip)
-					inst.VIPConfigs[index].VIP = ip
-					inst.DHCPInterfaceIPv6 = ip
-					if !p.config.DisableServiceUpdates {
-						if err := p.updateStatus(ctx, inst); err != nil {
-							log.Warn("updating svc", "err", err)
+				for {
+					select {
+					case <-ctx.Done():
+						log.Debug("IPv6 update watcher stopping")
+						return
+					case ip := <-inst.DHCPv6Client.IPChannel():
+						log.Debug("IP changed", "ip", ip)
+						inst.VIPConfigs[index].VIP = ip
+						inst.DHCPInterfaceIPv6 = ip
+						if !p.config.DisableServiceUpdates {
+							if err := p.updateStatus(ctx, inst); err != nil {
+								log.Warn("updating svc", "err", err)
+							}
 						}
 					}
 				}
-				log.Debug("IPv6 update channel closed, stopping")
 			}
 		})
 	}
--- a/pkg/vip/address.go
+++ b/pkg/vip/address.go
@@ -35,6 +35,9 @@ import (
 const (
 	defaultValidLft = 60
 	iptablesComment = "%s kube-vip load balancer IP"
+	// Linux reserves IFA_PROTO values 0-3 for unspecified and kernel-generated
+	// addresses; user-space protocols start at 4. This is only a safety floor,
+	firstUserAddressProtocol = 4

 	DefaultMaskIPv4 = 32
 	DefaultMaskIPv6 = 128
@@ -77,8 +80,7 @@ type Network interface {

 // network - This allows network configuration
 type network struct {
-	mu sync.Mutex
-
+	mu             sync.RWMutex
 	address        *netlink.Addr
 	link           *networkinterface.Link
 	ports          []v1.ServicePort
@@ -168,6 +170,7 @@ func NewConfig(address string, iface string, loGlobalScope bool, subnet string,
 		if err != nil {
 			return networks, errors.Wrapf(err, "could not parse address '%s'", address)
 		}
+		markKubeVIPAddress(result.address, result.routingProtocol)

 		// set address as deprecated so it isn't used as source address according to RFC 3484
 		result.address.PreferedLft = 0
@@ -265,6 +268,7 @@ func NewConfig(address string, iface string, loGlobalScope bool, subnet string,
 			if result.address, err = netlink.ParseAddr(fmt.Sprintf("%s/%s", ip, s)); err != nil {
 				return networks, err
 			}
+			markKubeVIPAddress(result.address, result.routingProtocol)
 			// set ValidLft so that the VIP expires if the DNS entry is updated, otherwise it'll be refreshed by the DNS prober
 			result.address.ValidLft = defaultValidLft

@@ -310,7 +314,17 @@ func ListRoutesByDst(table int, dst *net.IPNet) ([]netlink.Route, error) {
 	return routes, nil
 }

-func (configurator *network) PrepareRoute() *netlink.Route {
+func (configurator *network) PrepareRoute() (route *netlink.Route) {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
+	_ = configurator.link.WithInterface(func(intf netlink.Link) error {
+		route = configurator.prepareRoute(intf)
+		return nil
+	})
+	return route
+}
+
+func (configurator *network) prepareRoute(intf netlink.Link) *netlink.Route {
 	routeScope := netlink.SCOPE_UNIVERSE
 	if configurator.routingTableType == unix.RTN_LOCAL {
 		routeScope = netlink.SCOPE_LINK
@@ -318,7 +332,7 @@ func (configurator *network) PrepareRoute() *netlink.Route {
 	route := &netlink.Route{
 		Scope:     routeScope,
 		Dst:       configurator.address.IPNet,
-		LinkIndex: configurator.link.Intf.Attrs().Index,
+		LinkIndex: intf.Attrs().Index,
 		Table:     configurator.routeTable,
 		Type:      configurator.routingTableType,
 		Protocol:  netlink.RouteProtocol(configurator.routingProtocol),
@@ -339,32 +353,33 @@ func NetlinkHash(r *netlink.Route) string {
 }

 // AddRoute - Add an IP address to a route table
-func (configurator *network) AddRoute(precheck bool) (bool, error) {
-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
-	route := configurator.PrepareRoute()
-
-	exists := false
-	var err error
-	if precheck {
-		exists, err = configurator.routeExists(route)
-		if err != nil {
-			return false, errors.Wrap(err, "failed to check route")
+func (configurator *network) AddRoute(precheck bool) (added bool, err error) {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
+	err = configurator.link.WithInterface(func(intf netlink.Link) error {
+		route := configurator.prepareRoute(intf)
+		exists := false
+		if precheck {
+			var existsErr error
+			exists, existsErr = configurator.routeExists(intf, route)
+			if existsErr != nil {
+				return errors.Wrap(existsErr, "failed to check route")
+			}
 		}
-	}
-
-	if !exists {
-		if err := netlink.RouteAdd(route); err != nil {
-			return false, errors.Wrap(err, "failed to add route")
+		if exists {
+			return nil
 		}
-		return true, nil
-	}
-
-	return false, nil
+		if routeErr := netlink.RouteAdd(route); routeErr != nil {
+			return errors.Wrap(routeErr, "failed to add route")
+		}
+		added = true
+		return nil
+	})
+	return added, err
 }

-func (configurator *network) routeExists(route *netlink.Route) (bool, error) {
-	routes, err := netlink.RouteList(configurator.link.Intf, netlink.FAMILY_ALL)
+func (configurator *network) routeExists(intf netlink.Link, route *netlink.Route) (bool, error) {
+	routes, err := netlink.RouteList(intf, netlink.FAMILY_ALL)
 	if err != nil {
 		return false, errors.Wrap(err, "failed to list routes")
 	}
@@ -385,16 +400,18 @@ func (configurator *network) routeExists(route *netlink.Route) (bool, error) {
 // the same-prefix interface address is processed in the same netlink batch,
 // leaving the route in the kernel but never redistributed.
 func (configurator *network) ReplaceRoute() error {
-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
-	route := configurator.PrepareRoute()
-	configurator.reassertToggle = !configurator.reassertToggle
-	if configurator.reassertToggle {
-		route.Realm = 1
-	} else {
-		route.Realm = 2
-	}
-	return netlink.RouteReplace(route)
+	configurator.mu.Lock()
+	defer configurator.mu.Unlock()
+	return configurator.link.WithInterface(func(intf netlink.Link) error {
+		route := configurator.prepareRoute(intf)
+		configurator.reassertToggle = !configurator.reassertToggle
+		if configurator.reassertToggle {
+			route.Realm = 1
+		} else {
+			route.Realm = 2
+		}
+		return netlink.RouteReplace(route)
+	})
 }

 // DeleteRoute - Delete an IP address from a route table
@@ -405,6 +422,8 @@ func (configurator *network) DeleteRoute() error {

 // GetRoutes - Get an IP addresses from a route table
 func (configurator *network) getRoutes() (*[]netlink.Route, error) {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	routes, err := ListRoutesByDst(configurator.routeTable, configurator.address.IPNet)
 	if err != nil {
 		return nil, fmt.Errorf("error getting routes: %w", err)
@@ -443,12 +462,22 @@ func (configurator *network) shouldSkipDAD(override bool) bool {
 // precheck: if true, check if the IP already exists before adding
 // skipDAD: if true, set IFA_F_NODAD flag for IPv6 addresses to skip Duplicate Address Detection
 func (configurator *network) AddIP(precheck bool, skipDAD bool, minLifetime ...int) (bool, error) {
-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
+	configurator.mu.Lock()
+	defer configurator.mu.Unlock()
+	var added bool
+	err := configurator.link.WithInterface(func(intf netlink.Link) error {
+		var addErr error
+		added, addErr = configurator.addIP(intf, precheck, skipDAD, minLifetime...)
+		return addErr
+	})
+	return added, err
+}
+
+func (configurator *network) addIP(intf netlink.Link, precheck bool, skipDAD bool, minLifetime ...int) (bool, error) {
 	var existing *netlink.Addr
 	var err error
 	if precheck {
-		if existing, err = configurator.IsSet(); err != nil {
+		if existing, err = configurator.isSet(intf); err != nil {
 			return false, errors.Wrap(err, "could not check if address exists")
 		}
 	}
@@ -479,8 +508,8 @@ func (configurator *network) AddIP(precheck bool, skipDAD bool, minLifetime ...i
 	}

 	log.Debug("replacing IP", "address", configurator.address)
-	if err := netlink.AddrReplace(configurator.link.Intf, configurator.address); err != nil {
-		return false, errors.Wrap(err, fmt.Sprintf("could not add ip to device %q", configurator.link.Intf.Attrs().Name))
+	if err := netlink.AddrReplace(intf, configurator.address); err != nil {
+		return false, errors.Wrap(err, fmt.Sprintf("could not add ip to device %q", intf.Attrs().Name))
 	}

 	if configurator.nftables {
@@ -518,7 +547,7 @@ func (configurator *network) configureNFTables() error {
 		"ports", configurator.ports, "service-name", configurator.serviceName)

 	opt := nftables.TableFamilyIPv4
-	if utils.IsIPv6(configurator.IP()) {
+	if utils.IsIPv6(configurator.address.IP.String()) {
 		opt = nftables.TableFamilyIPv6
 	}

@@ -595,12 +624,13 @@ func (configurator *network) addNftablesRulesToLimitTrafficPorts(c *nfinternal.C
 		}
 	}

-	firstRule, err := insertCommonNFTablesRules(c, configurator.IP(), comment)
+	vip := configurator.address.IP.String()
+	firstRule, err := insertCommonNFTablesRules(c, vip, comment)
 	if err != nil {
 		return fmt.Errorf("could not add common nftables rules: %w", err)
 	}

-	if err := configurator.insertNFTablesRulesForServicePorts(c, configurator.IP(), comment, firstRule.Handle); err != nil {
+	if err := configurator.insertNFTablesRulesForServicePorts(c, vip, comment, firstRule.Handle); err != nil {
 		return fmt.Errorf("could not add nftables rules for service ports: %v", err)
 	}

@@ -1057,10 +1087,19 @@ func (configurator *network) removeNftablesRuleToLimitTrafficPorts(c *nfinternal

 // DeleteIP - Remove an IP address from the interface
 func (configurator *network) DeleteIP() (bool, error) {
-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
+	configurator.mu.Lock()
+	defer configurator.mu.Unlock()
+	var deleted bool
+	err := configurator.link.WithInterface(func(intf netlink.Link) error {
+		var deleteErr error
+		deleted, deleteErr = configurator.deleteIP(intf)
+		return deleteErr
+	})
+	return deleted, err
+}

-	result, err := configurator.IsSet()
+func (configurator *network) deleteIP(intf netlink.Link) (bool, error) {
+	result, err := configurator.isSet(intf)
 	if err != nil {
 		return false, errors.Wrap(err, "ip check in DeleteIP failed")
 	}
@@ -1070,7 +1109,7 @@ func (configurator *network) DeleteIP() (bool, error) {
 		return false, nil
 	}

-	if err = netlink.AddrDel(configurator.link.Intf, configurator.address); err != nil {
+	if err = netlink.AddrDel(intf, configurator.address); err != nil {
 		return false, errors.Wrap(err, "could not delete ip")
 	}

@@ -1140,7 +1179,8 @@ func (configurator *network) addIptablesRulesForMasquerade() error {

 // TO DO: It seems it is not be possible to use google/nftables with IPVS due to lack of IPVS matcher in nft
 func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client, comment string) error {
-	cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.IP(), configurator.IPVSMark())
+	vip := configurator.address.IP.String()
+	cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, vip, configurator.ipvsMark)

 	markChain := &nftables.Chain{
 		Name:     "ipvs_prerouting",
@@ -1154,7 +1194,7 @@ func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client,

 	markChain = c.AddChain(markChain)

-	ip := net.ParseIP(configurator.IP())
+	ip := net.ParseIP(vip)

 	if ip.To4() != nil {
 		ip = ip.To4()
@@ -1164,7 +1204,7 @@ func (configurator *network) addNftablesRulesForMasquerade(c *nfinternal.Client,

 	port := binaryutil.BigEndian.PutUint16(configurator.ipvsPort)

-	mark := binaryutil.NativeEndian.PutUint32(configurator.IPVSMark())
+	mark := binaryutil.NativeEndian.PutUint32(configurator.ipvsMark)

 	markRule := &nftables.Rule{
 		Table:    markChain.Table,
@@ -1286,7 +1326,7 @@ func (configurator *network) removeNftablesRulesForMasquerade(c *nfinternal.Clie
 	if configurator.serviceName != "" {
 		comment = fmt.Sprintf(iptablesComment, configurator.serviceName)
 	}
-	cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.IP(), configurator.IPVSMark())
+	cmt := fmt.Sprintf("%s - IPVS, VIP %s, MARK %d", comment, configurator.address.IP.String(), configurator.ipvsMark)

 	r, err := c.FindRuleByComment(chain.Table, chain, cmt)
 	if err != nil {
@@ -1349,27 +1389,26 @@ func delMasqueradeRuleForVIP(ipt *iptables.IPTables, vip, comment string) error

 // IsDADFAIL - Returns true if the address is IPv6 and has DADFAILED flag
 func (configurator *network) IsDADFAIL() bool {
-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
-
-	if configurator.address == nil || !utils.IsIPv6(configurator.address.IP.String()) {
-		return false
-	}
-
-	// Get all the address
-	addresses, err := netlink.AddrList(configurator.link.Intf, netlink.FAMILY_V6)
-	if err != nil {
-		return false
-	}
-
-	// Find the VIP and check if it is DADFAILED
-	for _, address := range addresses {
-		if address.IP.Equal(configurator.address.IP) && addressHasDADFAILEDFlag(address) {
-			return true
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
+	var dadFailed bool
+	_ = configurator.link.WithInterface(func(intf netlink.Link) error {
+		if configurator.address == nil || !utils.IsIPv6(configurator.address.IP.String()) {
+			return nil
 		}
-	}
-
-	return false
+		addresses, err := netlink.AddrList(intf, netlink.FAMILY_V6)
+		if err != nil {
+			return nil
+		}
+		for _, address := range addresses {
+			if address.IP.Equal(configurator.address.IP) && addressHasDADFAILEDFlag(address) {
+				dadFailed = true
+				break
+			}
+		}
+		return nil
+	})
+	return dadFailed
 }

 func addressHasDADFAILEDFlag(address netlink.Addr) bool {
@@ -1378,6 +1417,16 @@ func addressHasDADFAILEDFlag(address netlink.Addr) bool {

 // isSet - Check to see if VIP is set
 func (configurator *network) IsSet() (result *netlink.Addr, err error) {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
+	err = configurator.link.WithInterface(func(intf netlink.Link) error {
+		result, err = configurator.isSet(intf)
+		return err
+	})
+	return result, err
+}
+
+func (configurator *network) isSet(intf netlink.Link) (result *netlink.Addr, err error) {
 	var addresses []netlink.Addr

 	if configurator.address == nil {
@@ -1388,7 +1437,7 @@ func (configurator *network) IsSet() (result *netlink.Addr, err error) {
 		return nil, nil
 	}

-	addresses, err = netlink.AddrList(configurator.link.Intf, 0)
+	addresses, err = netlink.AddrList(intf, 0)
 	if err != nil {
 		err = errors.Wrap(err, "could not list addresses")

@@ -1408,10 +1457,10 @@ func (configurator *network) IsSet() (result *netlink.Addr, err error) {
 func (configurator *network) SetIP(ip string) error {
 	configurator.mu.Lock()
 	defer configurator.mu.Unlock()
+	return configurator.setIP(ip)
+}

-	configurator.link.Lock.Lock()
-	defer configurator.link.Lock.Unlock()
-
+func (configurator *network) setIP(ip string) error {
 	if strings.Contains("/", ip) {
 		return fmt.Errorf("ip should not contain CIDR notation got: %s", ip)
 	}
@@ -1443,7 +1492,7 @@ func (configurator *network) SetIP(ip string) error {
 	if err != nil {
 		return err
 	}
-	if configurator.address != nil && configurator.IsDNS() {
+	if configurator.address != nil && configurator.dnsName != "" {
 		addr.ValidLft = defaultValidLft
 	} else {
 		addr.ValidLft = math.MaxInt
@@ -1453,9 +1502,94 @@ func (configurator *network) SetIP(ip string) error {
 	addr.PreferedLft = 0

 	configurator.address = addr
+	if configurator.routingProtocol != 0 {
+		markKubeVIPAddress(configurator.address, configurator.routingProtocol)
+	}
 	return nil
 }

+// IsKubeVIPAddress reports whether an address carries kube-vip's configured
+// protocol origin. IFA_PROTO is supported for both IPv4 and IPv6 on Linux 5.18+.
+func IsKubeVIPAddress(address netlink.Addr, protocol int) bool {
+	return protocol >= firstUserAddressProtocol && address.Protocol == protocol
+}
+
+// RetainedKubeVIPAddressKeys returns the tagged kernel addresses matching the
+// supplied VIPs. Callers use the keys with CleanupKubeVIPAddresses after they
+// determine which VIPs remain locally referenced.
+func RetainedKubeVIPAddressKeys(protocol int, vips map[string]struct{}) (map[string]struct{}, error) {
+	retained := make(map[string]struct{})
+	err := forEachKubeVIPAddress(protocol, func(_ netlink.Link, address netlink.Addr) error {
+		if _, retain := vips[address.IP.String()]; retain {
+			retained[addressKey(address)] = struct{}{}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return retained, nil
+}
+
+// CleanupKubeVIPAddresses removes kube-vip addresses not reasserted by this
+// process. The retained keys must come from successful AddrReplace operations.
+func CleanupKubeVIPAddresses(protocol int, retained map[string]struct{}) (int, error) {
+	removed := 0
+	err := forEachKubeVIPAddress(protocol, func(link netlink.Link, address netlink.Addr) error {
+		key := addressKey(address)
+		if key == "" {
+			return nil
+		}
+		if _, keep := retained[key]; keep {
+			return nil
+		}
+		if err := netlink.AddrDel(link, &address); err != nil {
+			return errors.Wrapf(err, "deleting kube-vip address %q from interface %q", address.IP, link.Attrs().Name)
+		}
+		removed++
+		return nil
+	})
+	if err != nil {
+		return removed, err
+	}
+	return removed, nil
+}
+
+func forEachKubeVIPAddress(protocol int, visit func(netlink.Link, netlink.Addr) error) error {
+	links, err := netlink.LinkList()
+	if err != nil {
+		return errors.Wrap(err, "listing network links")
+	}
+	for _, link := range links {
+		addresses, err := netlink.AddrList(link, netlink.FAMILY_ALL)
+		if err != nil {
+			return errors.Wrapf(err, "listing addresses on interface %q", link.Attrs().Name)
+		}
+		for _, address := range addresses {
+			if IsKubeVIPAddress(address, protocol) {
+				if err := visit(link, address); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func addressKey(address netlink.Addr) string {
+	if address.LinkIndex <= 0 || address.IP == nil {
+		return ""
+	}
+	prefixLength, _ := address.Mask.Size()
+	return fmt.Sprintf("%d/%s/%d", address.LinkIndex, address.IP, prefixLength)
+}
+
+func markKubeVIPAddress(address *netlink.Addr, protocol int) {
+	if address != nil {
+		address.Protocol = protocol
+	}
+}
+
 // SetServicePorts updates the service ports from the service
 // If you want to limit traffic to the VIP to only the service ports, add service ports to the network firstly.
 func (configurator *network) SetServicePorts(service *v1.Service) {
@@ -1469,8 +1603,8 @@ func (configurator *network) SetServicePorts(service *v1.Service) {

 // IP - return the IP Address
 func (configurator *network) IP() string {
-	configurator.mu.Lock()
-	defer configurator.mu.Unlock()
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()

 	if configurator.address == nil || configurator.address.IP == nil {
 		return ""
@@ -1480,8 +1614,8 @@ func (configurator *network) IP() string {
 }

 func (configurator *network) CIDR() string {
-	configurator.mu.Lock()
-	defer configurator.mu.Unlock()
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()

 	if configurator.address == nil || configurator.address.IPNet == nil {
 		return ""
@@ -1492,24 +1626,30 @@ func (configurator *network) CIDR() string {

 // IP - return the IP Address
 func (configurator *network) IPisLinkLocal() bool {
-	configurator.mu.Lock()
-	defer configurator.mu.Unlock()
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()

 	return configurator.address.IP.IsLinkLocalUnicast()
 }

 // DNSName return the configured dnsName when use DNS
 func (configurator *network) DNSName() string {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.dnsName
 }

 // IsDNS - when dnsName is configured
 func (configurator *network) IsDNS() bool {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.dnsName != ""
 }

 // IsDDNS - return true if use dynamic dns
 func (configurator *network) IsDDNS() bool {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.isDDNS
 }

@@ -1518,12 +1658,19 @@ func (configurator *network) IsDDNS() bool {
 // it's expected that dynamic DNS should be configured so
 // the fqdn for apiserver endpoint is dDNSHostName.{LocalDomain}
 func (configurator *network) DDNSHostName() string {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return getHostName(configurator.dnsName)
 }

 // Interface - return the Interface name
 func (configurator *network) Interface() string {
-	return configurator.link.Intf.Attrs().Name
+	var name string
+	_ = configurator.link.WithInterface(func(intf netlink.Link) error {
+		name = intf.Attrs().Name
+		return nil
+	})
+	return name
 }

 func GarbageCollect(adapter, address string, intfMgr *networkinterface.Manager) (found bool, err error) {
@@ -1534,32 +1681,28 @@ func GarbageCollect(adapter, address string, intfMgr *networkinterface.Manager)
 	}

 	l := intfMgr.Get(link)
-
-	l.Lock.Lock()
-	defer l.Lock.Unlock()
-
-	// Get addresses on adapter
-	addrs, err := netlink.AddrList(l.Intf, netlink.FAMILY_ALL)
-	if err != nil {
-		return false, err
-	}
-
-	// Compare all addresses to new service address, and remove if needed
-	for _, existing := range addrs {
-		if existing.IP.String() == address {
-			// We've found the existing address
-			found = true
-			// linting issue
-			existing := existing
-			if err = netlink.AddrDel(l.Intf, &existing); err != nil {
-				return true, errors.Wrap(err, "could not delete ip")
+	err = l.WithInterface(func(intf netlink.Link) error {
+		addrs, listErr := netlink.AddrList(intf, netlink.FAMILY_ALL)
+		if listErr != nil {
+			return listErr
+		}
+		for _, existing := range addrs {
+			if existing.IP.String() == address {
+				found = true
+				existing := existing
+				if deleteErr := netlink.AddrDel(intf, &existing); deleteErr != nil {
+					return errors.Wrap(deleteErr, "could not delete ip")
+				}
 			}
 		}
-	}
-	return // Didn't find the address on the adapter
+		return nil
+	})
+	return found, err
 }

 func (configurator *network) SetMask(mask string) error {
+	configurator.mu.Lock()
+	defer configurator.mu.Unlock()
 	selectedMask := mask
 	var err error

@@ -1567,8 +1710,12 @@ func (configurator *network) SetMask(mask string) error {
 		return fmt.Errorf("no mask provided")
 	}

-	if configurator.IP() != "" {
-		selectedMask, err = SelectSubnet(configurator.IP(), mask)
+	ip := ""
+	if configurator.address != nil && configurator.address.IP != nil {
+		ip = configurator.address.IP.String()
+	}
+	if ip != "" {
+		selectedMask, err = SelectSubnet(ip, mask)
 		if err != nil {
 			return fmt.Errorf("failed to select mask %q: %w", mask, err)
 		}
@@ -1584,8 +1731,8 @@ func (configurator *network) SetMask(mask string) error {
 	size := DefaultMaskIPv4
 	family := utils.IPv4Family

-	if configurator.IP() != "" {
-		if utils.IsIPv6(configurator.IP()) {
+	if ip != "" {
+		if utils.IsIPv6(ip) {
 			size = DefaultMaskIPv6
 			family = utils.IPv6Family
 		}
@@ -1600,20 +1747,29 @@ func (configurator *network) SetMask(mask string) error {
 		return fmt.Errorf("failed to create mask /%d", m)
 	}

-	configurator.mu.Lock()
-	defer configurator.mu.Unlock()
-
 	configurator.address.Mask = toSet
 	return nil
 }

 func (configurator *network) SetHasEndpoints(value bool) {
-	log.Debug("setting HasEndpoints", "ip", configurator.IP(), "value", value)
+	configurator.mu.Lock()
+	defer configurator.mu.Unlock()
+	ip := ""
+	if configurator.address != nil && configurator.address.IP != nil {
+		ip = configurator.address.IP.String()
+	}
+	log.Debug("setting HasEndpoints", "ip", ip, "value", value)
 	configurator.hasEndpoints = value
 }

 func (configurator *network) HasEndpoints() bool {
-	log.Debug("getting HasEndpoints", "ip", configurator.IP(), "value", configurator.hasEndpoints)
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
+	ip := ""
+	if configurator.address != nil && configurator.address.IP != nil {
+		ip = configurator.address.IP.String()
+	}
+	log.Debug("getting HasEndpoints", "ip", ip, "value", configurator.hasEndpoints)
 	return configurator.hasEndpoints
 }

@@ -1622,14 +1778,20 @@ func (configurator *network) ARPName() string {
 }

 func (configurator *network) GetPossibleSubnets() string {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.possibleSubnets
 }

 func (configurator *network) DHCPFamily() string {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.dhcpFamily
 }

 func (configurator *network) IPVSMark() uint32 {
+	configurator.mu.RLock()
+	defer configurator.mu.RUnlock()
 	return configurator.ipvsMark
 }

--- a/pkg/vip/address_dad_linux_test.go
+++ b/pkg/vip/address_dad_linux_test.go
@@ -16,11 +16,10 @@ func TestAddIPPerCallDADSkipDoesNotPersist(t *testing.T) {
 		t.Fatal(err)
 	}

+	interfaceManager := networkinterface.NewManager()
 	configurator := &network{
 		address: address,
-		link: &networkinterface.Link{
-			Intf: &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kube-vip-dad-test"}},
-		},
+		link:    interfaceManager.Get(&netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kube-vip-dad-test"}}),
 	}

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

 import (
 	"testing"
+
+	"github.com/kube-vip/kube-vip/pkg/networkinterface"
 )

+func TestNewConfigTagsStaticAndUpdatedAddresses(t *testing.T) {
+	const protocol = 248
+	networks, err := NewConfig("192.0.2.10", "lo", false, "32", false, "", false, false, 0, 0, protocol,
+		"", "", "", false, 0, false, networkinterface.NewManager(), false, false)
+	if err != nil {
+		t.Fatalf("NewConfig() error = %v", err)
+	}
+	configured, ok := networks[0].(*network)
+	if !ok {
+		t.Fatalf("network type = %T, want *network", networks[0])
+	}
+	if configured.address.Protocol != protocol {
+		t.Fatalf("static address protocol = %d, want %d", configured.address.Protocol, protocol)
+	}
+	if err := configured.SetIP("192.0.2.11"); err != nil {
+		t.Fatalf("SetIP() error = %v", err)
+	}
+	if configured.address.Protocol != protocol {
+		t.Fatalf("updated address protocol = %d, want %d", configured.address.Protocol, protocol)
+	}
+}
+
 func TestShouldSkipDAD(t *testing.T) {
 	cases := []struct {
 		name     string
--- a/pkg/vip/dhcpv4.go
+++ b/pkg/vip/dhcpv4.go
@@ -73,7 +73,6 @@ func (c *DHCPv4Client) Stop() {

 func (c *DHCPv4Client) close() {
 	c.stopOnce.Do(func() {
-		close(c.ipChan)
 		close(c.stopChan)
 	})
 }
@@ -284,7 +283,12 @@ RequestLoop:
 	}

 	if c.ipChan != nil {
-		c.ipChan <- lease.ACK.YourIPAddr.String()
+		// Nothing closes ipChan, so never block on a consumer that already stopped.
+		select {
+		case c.ipChan <- lease.ACK.YourIPAddr.String():
+		case <-c.stopChan:
+		case <-ctx.Done():
+		}
 	}

 	return lease, nil
--- a/pkg/vip/dhcpv6.go
+++ b/pkg/vip/dhcpv6.go
@@ -23,6 +23,9 @@ func init() {
 }

 type DHCPv6ClientManager struct {
+	// mu guards clients and the reference counts of its entries together, so a
+	// concurrent Add cannot join a client that Delete is already retiring.
+	mu      sync.Mutex
 	clients map[string]*DHCPv6InternalClient
 }

@@ -33,17 +36,17 @@ func NewDHCPv6ClientManager() *DHCPv6ClientManager {
 }

 func (m *DHCPv6ClientManager) Get(iface string) *DHCPv6InternalClient {
-	c, exists := m.clients[iface]
-	if !exists {
-		return nil
-	}
-	return c
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	return m.clients[iface]
 }

 func (m *DHCPv6ClientManager) Add(iface string) (*DHCPv6InternalClient, error) {
-	c := m.Get(iface)
+	m.mu.Lock()
+	defer m.mu.Unlock()

-	if c != nil {
+	if c := m.clients[iface]; c != nil {
 		c.references.Add(1)
 		return c, nil
 	}
@@ -57,15 +60,16 @@ func (m *DHCPv6ClientManager) Add(iface string) (*DHCPv6InternalClient, error) {
 }

 func (m *DHCPv6ClientManager) Delete(iface string) {
-	c := m.Get(iface)
+	m.mu.Lock()
+	defer m.mu.Unlock()

-	if c != nil {
-		c.references.Add(-1)
-		ref := c.references.Load()
-		if ref < 1 {
-			c.client.Close()
-			delete(m.clients, iface)
-		}
+	c := m.clients[iface]
+	if c == nil {
+		return
+	}
+	if c.references.Add(-1) < 1 {
+		c.client.Close()
+		delete(m.clients, iface)
 	}
 }

@@ -154,7 +158,6 @@ func (c *DHCPv6Client) Stop() {
 // Close dhcp client channels
 func (c *DHCPv6Client) close() {
 	c.stop.Do(func() {
-		close(c.ipChan)
 		close(c.stopChan)
 	})
 	dhcpv6ClientManager.Delete(c.managerKey)
@@ -304,7 +307,12 @@ RequestLoop:
 	}

 	if c.ipChan != nil {
-		c.ipChan <- addr.IPv6Addr.String()
+		// Nothing closes ipChan, so never block on a consumer that already stopped.
+		select {
+		case c.ipChan <- addr.IPv6Addr.String():
+		case <-c.stopChan:
+		case <-ctx.Done():
+		}
 	}

 	return addr, nil
--- a/pkg/vip/dhcpv6_robustness_test.go
+++ b/pkg/vip/dhcpv6_robustness_test.go
@@ -2,6 +2,7 @@ package vip

 import (
 	"net"
+	"sync"
 	"sync/atomic"
 	"testing"

@@ -36,6 +37,38 @@ func TestDHCPv6StopReleasesManagerReferenceForParentInterface(t *testing.T) {
 	}
 }

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

 	log "log/slog"

@@ -14,6 +15,8 @@ import (
 	"github.com/vishvananda/netlink"
 )

+var errDefaultInterfaceSubscriptionClosed = errors.New("default interface subscription closed")
+
 // getHostName return the hostname from the fqdn
 func getHostName(dnsName string) string {
 	if dnsName == "" {
@@ -75,26 +78,123 @@ func getDefaultRoute(family int) (*net.Interface, error) {
 	return nil, errors.New("default route not found")
 }

-// MonitorDefaultInterface monitor the default interface and catch the event of the default route
+// MonitorDefaultInterface monitors the default interface for route removal or link loss.
 func MonitorDefaultInterface(ctx context.Context, defaultIF *net.Interface) error {
-	routeCh := make(chan netlink.RouteUpdate)
+	return monitorDefaultInterfaceWithRetry(ctx, defaultIF, subscribeDefaultInterface, GetDefaultGatewayInterface, 100*time.Millisecond)
+}
+
+func monitorDefaultInterfaceWithRetry(ctx context.Context, defaultIF *net.Interface,
+	subscribe func(context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error),
+	lookup func() (*net.Interface, error), retryDelay time.Duration) error {
+	for {
+		monitorCtx, cancel := context.WithCancel(ctx)
+		routeCh, linkCh, err := subscribe(monitorCtx)
+		if err == nil {
+			err = monitorDefaultInterface(monitorCtx, defaultIF, routeCh, linkCh)
+		}
+		cancel()
+		drainDefaultInterfaceSubscriptions(routeCh, linkCh)
+		if ctx.Err() != nil {
+			return nil
+		}
+		if err != nil && !errors.Is(err, errDefaultInterfaceSubscriptionClosed) {
+			log.Warn("default interface subscription failed, retrying", "err", err)
+		} else if err == nil {
+			return nil
+		}
+		if refreshed, lookupErr := lookup(); lookupErr == nil {
+			defaultIF = refreshed
+		} else {
+			log.Warn("failed to refresh default interface while resubscribing", "err", lookupErr)
+		}
+		timer := time.NewTimer(retryDelay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil
+		case <-timer.C:
+		}
+	}
+}
+
+func subscribeDefaultInterface(ctx context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error) {
+	const subscriptionBuffer = 64
+	routeCh := make(chan netlink.RouteUpdate, subscriptionBuffer)
 	if err := netlink.RouteSubscribe(routeCh, ctx.Done()); err != nil {
-		return fmt.Errorf("subscribe route failed, error: %w", err)
+		return nil, nil, fmt.Errorf("subscribe route failed, error: %w", err)
+	}
+	linkCh := make(chan netlink.LinkUpdate, subscriptionBuffer)
+	if err := netlink.LinkSubscribe(linkCh, ctx.Done()); err != nil {
+		return routeCh, nil, fmt.Errorf("subscribe link failed, error: %w", err)
 	}

+	return routeCh, linkCh, nil
+}
+
+func monitorDefaultInterface(ctx context.Context, defaultIF *net.Interface, routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) error {
 	for {
 		select {
-		case r := <-routeCh:
+		case r, ok := <-routeCh:
+			if !ok {
+				return subscriptionClosed(ctx, "route")
+			}
 			log.Debug(fmt.Sprintf("type: %d, route: %+v", r.Type, r.Route))
-			if r.Type == syscall.RTM_DELROUTE && (r.Dst == nil || r.Dst.String() == "0.0.0.0/0") && r.LinkIndex == defaultIF.Index {
+			if r.Type == syscall.RTM_DELROUTE && isDefaultRoute(r.Dst) && r.LinkIndex == defaultIF.Index {
 				return fmt.Errorf("default route deleted and the default interface may be invalid")
 			}
+		case update, ok := <-linkCh:
+			if !ok {
+				return subscriptionClosed(ctx, "link")
+			}
+			if update.Link == nil {
+				continue
+			}
+			attrs := update.Attrs()
+			if attrs != nil && attrs.Index == defaultIF.Index && attrs.Flags&net.FlagUp == 0 {
+				return fmt.Errorf("default interface %q is down", defaultIF.Name)
+			}
 		case <-ctx.Done():
 			return nil
 		}
 	}
 }

+func subscriptionClosed(ctx context.Context, subscription string) error {
+	if ctx.Err() != nil {
+		return nil
+	}
+	return fmt.Errorf("%w: %s subscription closed", errDefaultInterfaceSubscriptionClosed, subscription)
+}
+
+// isDefaultRoute accepts both families, matching the selection made by
+// GetDefaultGatewayInterface.
+func isDefaultRoute(dst *net.IPNet) bool {
+	if dst == nil {
+		return true
+	}
+	return dst.String() == "0.0.0.0/0" || dst.String() == "::/0"
+}
+
+func drainDefaultInterfaceSubscriptions(routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) {
+	timer := time.NewTimer(100 * time.Millisecond)
+	defer timer.Stop()
+
+	for routeCh != nil || linkCh != nil {
+		select {
+		case _, ok := <-routeCh:
+			if !ok {
+				routeCh = nil
+			}
+		case _, ok := <-linkCh:
+			if !ok {
+				linkCh = nil
+			}
+		case <-timer.C:
+			return
+		}
+	}
+}
+
 func GenerateMac() (mac string) {
 	buf := make([]byte, 3)
 	_, err := rand.Read(buf)
--- a/pkg/vip/util_linux_test.go
+++ b/pkg/vip/util_linux_test.go
@@ -0,0 +1,293 @@
+//go:build linux
+
+package vip
+
+import (
+	"context"
+	"net"
+	"runtime"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/vishvananda/netlink"
+	"github.com/vishvananda/netns"
+)
+
+func TestMonitorDefaultInterfaceDetectsDefaultRouteDeletionPerFamily(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		cidr string
+	}{
+		{name: "IPv4", cidr: "0.0.0.0/0"},
+		{name: "IPv6", cidr: "::/0"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			defaultIF := &net.Interface{Index: 7, Name: "test0"}
+			_, defaultRoute, err := net.ParseCIDR(test.cidr)
+			if err != nil {
+				t.Fatalf("ParseCIDR() error = %v", err)
+			}
+			routeCh := make(chan netlink.RouteUpdate, 1)
+			routeCh <- netlink.RouteUpdate{
+				Type:  syscall.RTM_DELROUTE,
+				Route: netlink.Route{Dst: defaultRoute, LinkIndex: defaultIF.Index},
+			}
+			linkCh := make(chan netlink.LinkUpdate)
+
+			err = monitorDefaultInterfaceForTest(t, context.Background(), defaultIF, routeCh, linkCh)
+			if err == nil || !strings.Contains(err.Error(), "default route deleted") {
+				t.Fatalf("monitor error = %v, want a default route deletion error", err)
+			}
+		})
+	}
+}
+
+func TestMonitorDefaultInterfaceReturnsErrorWhenLinkGoesDown(t *testing.T) {
+	defaultIF := &net.Interface{Index: 7, Name: "test0"}
+	routeCh := make(chan netlink.RouteUpdate)
+	linkCh := make(chan netlink.LinkUpdate, 1)
+	linkCh <- netlink.LinkUpdate{
+		Link: &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Index: defaultIF.Index}},
+	}
+
+	err := monitorDefaultInterfaceForTest(t, context.Background(), defaultIF, routeCh, linkCh)
+	if err == nil {
+		t.Fatal("expected an error when the default interface goes down")
+	}
+	if !strings.Contains(err.Error(), "default interface \"test0\" is down") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestMonitorDefaultInterfaceHandlesClosedSubscriptions(t *testing.T) {
+	for _, test := range []struct {
+		name       string
+		closeRoute bool
+		cancel     bool
+		wantErr    string
+	}{
+		{name: "closed route subscription", closeRoute: true, wantErr: "route subscription closed"},
+		{name: "closed link subscription", wantErr: "link subscription closed"},
+		{name: "context cancellation with closed route subscription", closeRoute: true, cancel: true},
+		{name: "context cancellation with closed link subscription", cancel: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			routeCh := make(chan netlink.RouteUpdate)
+			linkCh := make(chan netlink.LinkUpdate)
+			if test.closeRoute {
+				close(routeCh)
+			} else {
+				close(linkCh)
+			}
+			if test.cancel {
+				cancel()
+			}
+
+			err := monitorDefaultInterfaceForTest(t, ctx, &net.Interface{}, routeCh, linkCh)
+			if test.wantErr == "" {
+				if err != nil {
+					t.Fatalf("monitor error = %v, want nil", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), test.wantErr) {
+				t.Fatalf("monitor error = %v, want %q", err, test.wantErr)
+			}
+		})
+	}
+}
+
+func monitorDefaultInterfaceForTest(t *testing.T, ctx context.Context, defaultIF *net.Interface,
+	routeCh <-chan netlink.RouteUpdate, linkCh <-chan netlink.LinkUpdate) error {
+	t.Helper()
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- monitorDefaultInterface(ctx, defaultIF, routeCh, linkCh)
+	}()
+	select {
+	case err := <-errCh:
+		return err
+	case <-time.After(time.Second):
+		t.Fatal("default interface monitor did not return")
+		return nil
+	}
+}
+
+func TestMonitorDefaultInterfaceHandlesClosedSubscriptionsAfterCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	routeCh := make(chan netlink.RouteUpdate)
+	linkCh := make(chan netlink.LinkUpdate)
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- monitorDefaultInterface(ctx, &net.Interface{}, routeCh, linkCh)
+	}()
+
+	cancel()
+	close(routeCh)
+	close(linkCh)
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("monitor error = %v, want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("monitor did not stop after cancellation")
+	}
+}
+
+func TestMonitorDefaultInterfaceRetriesClosedSubscription(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	secondSubscribed := make(chan struct{})
+	attempts := 0
+	subscribe := func(ctx context.Context) (chan netlink.RouteUpdate, chan netlink.LinkUpdate, error) {
+		if err := ctx.Err(); err != nil {
+			return nil, nil, err
+		}
+		attempts++
+		routeCh := make(chan netlink.RouteUpdate)
+		linkCh := make(chan netlink.LinkUpdate)
+		if attempts == 1 {
+			close(routeCh)
+			close(linkCh)
+			return routeCh, linkCh, nil
+		}
+		close(secondSubscribed)
+		go func() {
+			<-ctx.Done()
+			close(routeCh)
+			close(linkCh)
+		}()
+		return routeCh, linkCh, nil
+	}
+	lookup := func() (*net.Interface, error) {
+		return &net.Interface{Index: 2, Name: "refreshed"}, nil
+	}
+	done := make(chan error, 1)
+	go func() {
+		done <- monitorDefaultInterfaceWithRetry(ctx, &net.Interface{Index: 1, Name: "original"}, subscribe, lookup, time.Millisecond)
+	}()
+	select {
+	case <-secondSubscribed:
+	case <-time.After(time.Second):
+		t.Fatal("monitor did not resubscribe after channel closure")
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("monitor returned an error after cancellation: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("resubscribed monitor did not stop after cancellation")
+	}
+	if attempts != 2 {
+		t.Fatalf("subscription attempts = %d, want 2", attempts)
+	}
+}
+
+func TestMonitorDefaultInterfaceIgnoresNilLink(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	routeCh := make(chan netlink.RouteUpdate)
+	linkCh := make(chan netlink.LinkUpdate)
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- monitorDefaultInterface(ctx, &net.Interface{}, routeCh, linkCh)
+	}()
+
+	linkCh <- netlink.LinkUpdate{}
+	cancel()
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("monitor error = %v, want nil", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("monitor did not stop after nil link update")
+	}
+}
+
+func TestDrainDefaultInterfaceSubscriptionsIsBounded(t *testing.T) {
+	done := make(chan struct{})
+	go func() {
+		drainDefaultInterfaceSubscriptions(make(chan netlink.RouteUpdate), make(chan netlink.LinkUpdate))
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("subscription drain did not stop")
+	}
+}
+
+func TestMonitorDefaultInterfaceReturnsErrorWhenTestLinkIsSetDown(t *testing.T) {
+	runtime.LockOSThread()
+	defer runtime.UnlockOSThread()
+
+	originalNS, err := netns.Get()
+	if err != nil {
+		t.Fatalf("getting current network namespace: %v", err)
+	}
+	defer originalNS.Close()
+
+	testNS, err := netns.New()
+	if err != nil {
+		if requireNetworkNamespaces {
+			t.Fatalf("creating isolated network namespace: %v", err)
+		}
+		t.Skipf("creating isolated network namespace: %v", err)
+	}
+	defer testNS.Close()
+	defer func() {
+		if err := netns.Set(originalNS); err != nil {
+			t.Errorf("restoring network namespace: %v", err)
+		}
+	}()
+
+	link := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "kv-monitor0"}}
+	if err := netlink.LinkAdd(link); err != nil {
+		t.Fatalf("creating test interface: %v", err)
+	}
+	defer func() {
+		if err := netlink.LinkDel(link); err != nil {
+			t.Errorf("deleting test interface: %v", err)
+		}
+	}()
+	if err := netlink.LinkSetUp(link); err != nil {
+		t.Fatalf("bringing test interface up: %v", err)
+	}
+
+	defaultIF, err := net.InterfaceByName(link.Attrs().Name)
+	if err != nil {
+		t.Fatalf("getting test interface: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	routeCh, linkCh, err := subscribeDefaultInterface(ctx)
+	if err != nil {
+		t.Fatalf("subscribing to link updates: %v", err)
+	}
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- monitorDefaultInterface(ctx, defaultIF, routeCh, linkCh)
+	}()
+
+	if err := netlink.LinkSetDown(link); err != nil {
+		t.Fatalf("bringing test interface down: %v", err)
+	}
+
+	select {
+	case err := <-errCh:
+		if err == nil || !strings.Contains(err.Error(), "default interface \"kv-monitor0\" is down") {
+			t.Fatalf("monitor error = %v, want default-interface-down error", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("monitor did not report the interface going down")
+	}
+}
Author	SHA1	Message	Date
Marcel Fest	c925fb7cd9	fix: dhcp panic and leak potential Signed-off-by: Marcel Fest <marcel.fest@telekom.de>	2026-09-18 14:20:22 +02:00
Marcel Fest	d6129c8299	fix(vip): synchronize shared datapath state Serialize the interface link cache, route tracker and address configuration so concurrent Services cannot corrupt shared state or deadlock on nested address locks. Signed-off-by: Marcel Fest <marcel.fest@telekom.de>	2026-09-18 14:20:22 +02:00
Marcel Fest	3d73039cce	feat(lease): record VIP ownership in lease annotations (#1767 ) * feat(lease): record VIP ownership in lease annotations Encode the instance name, address protocol and an ordered VIP list on the Kubernetes lease so a restarting instance can tell which tagged addresses it still owns. Only the current holder publishes the annotation. Signed-off-by: Marcel Fest <marcel.fest@telekom.de> Signed-off-by: Cellebyte <marcel.fest@live.de> Signed-off-by: Marcel Fest <marcel.fest@telekom.de> * fix: flip flop on ordering when reading and writing the annotation Signed-off-by: Marcel Fest <marcel.fest@telekom.de> * fix: address feedback from review Signed-off-by: Marcel Fest <marcel.fest@telekom.de> --------- Signed-off-by: Marcel Fest <marcel.fest@telekom.de> Signed-off-by: Cellebyte <marcel.fest@live.de>	2026-09-18 14:20:19 +02:00
Patryk Strusiewicz-Surmacki	a60f80b1b2	Refactored and fixed metrics code Signed-off-by: Patryk Strusiewicz-Surmacki <patryk-pawel.strusiewicz-surmacki@external.telekom.de>	2026-09-16 18:01:07 +02:00