feat: lose leadership when interface is down

Signed-off-by: Bohdan Leshchenko <bohdan.leshchenko1@gmail.com>

cmd/kube-vip.go

@@ -71,6 +71,8 @@ func init() {
 	kubeVipCmd.PersistentFlags().BoolVar(&initConfig.EnableWireguard, "wireguard", false, "Enable Wireguard for services VIPs")
 	kubeVipCmd.PersistentFlags().BoolVar(&initConfig.EnableRoutingTable, "table", false, "Enable Routing Table for services VIPs")
 	kubeVipCmd.PersistentFlags().BoolVar(&initConfig.PreserveVIPOnLeadershipLoss, "preserveVipOnLeadershipLoss", false, "Preserve ARP VIP addresses on interface when leadership is lost (default: false for backward compatibility)")
+	kubeVipCmd.PersistentFlags().BoolVar(&initConfig.LoseLeadership, "loseLeadership", false, "Lose leadership when VIP interface goes down")
+	kubeVipCmd.PersistentFlags().IntVar(&initConfig.LoseLeadershipTimeoutSeconds, "loseLeadershiptTimeoutSeconds", 30, "Timeout before re-electing a leader when the VIP interface is down")
 
 	// LoadBalancer flags
 	kubeVipCmd.PersistentFlags().BoolVar(&initConfig.EnableLoadBalancer, "enableLoadBalancer", false, "enable loadbalancing on the VIP with IPVS")

pkg/arp/arp.go

@@ -10,6 +10,7 @@ import (
 	"github.com/kube-vip/kube-vip/pkg/kubevip"
 	"github.com/kube-vip/kube-vip/pkg/utils"
 	"github.com/kube-vip/kube-vip/pkg/vip"
+	"github.com/vishvananda/netlink"
 )
 
 type Manager struct {
@@ -113,7 +114,43 @@ func (m *Manager) Count(name string) int {
 	return 0
 }
 
-func (m *Manager) StartAdvertisement(ctx context.Context) {
+func (m *Manager) StartAdvertisement(ctx context.Context, killFunc func()) {
+	if m.config.LoseLeadership {
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		watchCtx, cancelWatch := context.WithCancel(ctx)
+		defer cancelWatch()
+
+		log.Info("[ARP manager] starting watching network device", "interface", m.config.Interface)
+
+		duration := time.Duration(m.config.LoseLeadershipTimeoutSeconds) * time.Second
+		timeout := time.NewTimer(duration)
+		timeout.Stop()
+
+		wg.Go(func() {
+			select {
+			case <-timeout.C:
+				killFunc()
+			case <-watchCtx.Done():
+				return
+			}
+		})
+
+		wg.Go(func() {
+			defer cancelWatch()
+			if err := watch(watchCtx, m.config.Interface, func(s netlink.LinkOperState) {
+				if isUp(s) {
+					timeout.Stop()
+					return
+				}
+				timeout.Reset(duration)
+			}); err != nil {
+				log.Warn("[ARP manager] stopped watching interface", "err", err)
+			}
+		})
+	}
+
 	log.Info("[ARP manager] starting ARP/NDP advertisement")
 
 	ticker := time.NewTicker(time.Duration(m.config.ArpBroadcastRate) * time.Millisecond)
@@ -200,7 +237,6 @@ func ensureIPAndSendGratuitous(instance *Instance) {
 				log.Warn(err.Error())
 			}
 		}
-
 	} else {
 		// Gratuitous ARP, will broadcast to new MAC <-> IPv4 address
 		err := vip.ARPSendGratuitous(ipString, iface)
@@ -209,3 +245,51 @@ func ensureIPAndSendGratuitous(instance *Instance) {
 		}
 	}
 }
+
+// watch subscribing to the network interface events and calls handler
+func watch(ctx context.Context, interfaceName string, operStateHandler func(netlink.LinkOperState)) error {
+	ifname, err := netlink.LinkByName(interfaceName)
+	if err != nil {
+		return fmt.Errorf("failed to watch interface %q: %w", interfaceName, err)
+	}
+
+	// verify if this interface is physical device
+	if _, ok := ifname.(*netlink.Device); !ok {
+		return fmt.Errorf("interface %s is not physical, ignoring", interfaceName)
+	}
+
+	events := make(chan netlink.LinkUpdate)
+	done := make(chan struct{})
+
+	if err := netlink.LinkSubscribe(events, done); err != nil {
+		return fmt.Errorf("failed to subscribe to the interface events: %w", err)
+	}
+	defer close(done)
+
+	//  handle initial state
+	operStateHandler(ifname.Attrs().OperState)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case event, ok := <-events:
+			if !ok {
+				return fmt.Errorf("interface events channel closed")
+			}
+
+			attrs := event.Attrs()
+			// LinkSubscribe captures events for all network devices found
+			// so we only care about vip interface
+			if ifname.Attrs().Name != attrs.Name {
+				continue
+			}
+			log.Debug("handling device change", "state", attrs.OperState)
+			operStateHandler(attrs.OperState)
+		}
+	}
+}
+
+func isUp(operState netlink.LinkOperState) bool {
+	return operState == netlink.OperUp
+}

pkg/kubevip/config_environment.go

@@ -47,6 +47,23 @@ func ParseEnvironment(c *Config) error {
 		c.LoInterfaceGlobalScope = b
 	}
 
+	env = os.Getenv(vipLoseLeadership)
+	if env != "" {
+		b, err := strconv.ParseBool(env)
+		if err != nil {
+			return err
+		}
+		c.LoseLeadership = b
+	}
+
+	env = os.Getenv(vipLoseLeadershipTimeoutSeconds)
+	if env != "" {
+		i, err := strconv.ParseInt(env, 10, 32)
+		if err != nil {
+			return fmt.Errorf("parsing env var %s (value: %s): %w", vipLoseLeadershipTimeoutSeconds, env, err)
+		}
+		c.LoseLeadershipTimeoutSeconds = int(i)
+	}
 	// Find (services) interface
 	env = os.Getenv(vipServicesInterface)
 	if env != "" {
@@ -995,6 +1012,10 @@ func mergeConfigValues(baseConfig, fileConfig *Config) {
 	if baseConfig.DebounceTime == debouncer.DefaultTime && fileConfig.DebounceTime != debouncer.DefaultTime {
 		baseConfig.DebounceTime = fileConfig.DebounceTime
 	}
+
+	if baseConfig.LoseLeadershipTimeoutSeconds == 0 && fileConfig.LoseLeadershipTimeoutSeconds != 0 {
+		baseConfig.LoseLeadershipTimeoutSeconds = fileConfig.LoseLeadershipTimeoutSeconds
+	}
 }
 
 // mergeBGPConfig merges BGP configuration

pkg/kubevip/config_envvar.go

@@ -36,6 +36,12 @@ const (
 	// vipInterface - defines the interface that the vip should bind too
 	vipInterface = "vip_interface"
 
+	// vipLoseLeadership - defines if leader should lose leadership if network interface is down
+	vipLoseLeadership = "vip_loseleadership"
+
+	// vipLoseLeadershipTimeout - defines the timeout for lose leadership
+	vipLoseLeadershipTimeoutSeconds = "vip_loseleadership_timeout_seconds"
+
 	// vipInterfaceLoGlobal - defines if the lo interface (if used) should have a global scope
 	vipInterfaceLoGlobal = "vip_interfaceloglobal"
 

pkg/kubevip/config_generator.go

@@ -390,6 +390,23 @@ func generatePodSpec(c *Config, image, imageVersion string, inCluster bool) (*co
 		newEnvironment = append(newEnvironment, leaderElection...)
 	}
 
+	if c.LoseLeadership {
+		loseLeadership := []corev1.EnvVar{
+			{
+				Name:  vipLoseLeadership,
+				Value: strconv.FormatBool(c.LoseLeadership),
+			},
+		}
+
+		if c.LoseLeadershipTimeoutSeconds > 0 {
+			loseLeadership = append(loseLeadership, corev1.EnvVar{
+				Name:  vipLoseLeadership,
+				Value: fmt.Sprintf("%d", c.LoseLeadershipTimeoutSeconds),
+			})
+		}
+		newEnvironment = append(newEnvironment, loseLeadership...)
+	}
+
 	// If we're enabling node labeling on leader election
 	if c.EnableNodeLabeling {
 		EnableNodeLabeling := []corev1.EnvVar{

pkg/kubevip/config_types.go

@@ -54,6 +54,12 @@ type Config struct {
 	// If false, VIP addresses are deleted on leadership loss (legacy behavior)
 	PreserveVIPOnLeadershipLoss bool `yaml:"preserveVipOnLeadershipLoss"`
 
+	// LoseLeadership enables leadership loss if VIP interface(physical) is down
+	LoseLeadership bool `yaml:"loseLeadership"`
+
+	// LoseLeadershipTimeoutSeconds defines the timeout after which interface will be considered down. Default is 30s
+	LoseLeadershipTimeoutSeconds int `yaml:"loseLeadershipTimeoutSeconds"`
+
 	// Annotations will define if we're going to wait and lookup configuration from Kubernetes node annotations
 	Annotations string
 

pkg/manager/worker/arp.go

@@ -25,7 +25,8 @@ func NewARP(arpMgr *arp.Manager, intfMgr *networkinterface.Manager,
 	config *kubevip.Config, closing *atomic.Bool, killFunc func(),
 	svcProcessor *services.Processor, mutex *sync.Mutex, clientSet *kubernetes.Clientset,
 	electionMgr *election.Manager, leaseMgr *lease.Manager, routeMgr *route.Manager,
-	nodeLabelMgr node.Labeler) *ARP {
+	nodeLabelMgr node.Labeler,
+) *ARP {
 	return &ARP{
 		Common: *newCommon(arpMgr, intfMgr, config, closing, killFunc,
 			svcProcessor, mutex, clientSet, electionMgr, leaseMgr, routeMgr,
@@ -36,7 +37,7 @@ func NewARP(arpMgr *arp.Manager, intfMgr *networkinterface.Manager,
 func (a *ARP) Configure(ctx context.Context, wg *sync.WaitGroup) error {
 	log.Info("Start ARP/NDP advertisement Global")
 	wg.Go(func() {
-		a.arpMgr.StartAdvertisement(ctx)
+		a.arpMgr.StartAdvertisement(ctx, a.killFunc)
 	})
 	return nil
 }
@@ -52,7 +53,6 @@ func (a *ARP) StartControlPlane(ctx context.Context, electionManager *election.M
 }
 
 func (a *ARP) ConfigureServices() {
-
 }
 
 func (a *ARP) StartServices(ctx context.Context) error {