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

Do not use panic()

Browse Source 
Signed-off-by: Patryk Strusiewicz-Surmacki <patryk.pawel.strusiewicz-surmacki@external.telekom.de>
This commit is contained in:
Patryk Strusiewicz-Surmacki
2025-12-31 15:21:23 +01:00
committed by Marcel Fest
parent 33c8bc08ac
commit 5818a6c661
20 changed files with 348 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
cmd/kube-vip-kubeadm.go
View File

@@ -65,7 +65,11 @@ var kubeKubeadmInit = &cobra.Command{
}
}
cfg := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
cfg, err := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
if err != nil {
log.Error("unable to create manifest", "err", err)
return
}
fmt.Println(cfg) // output manifest to stdout
},
}
@@ -110,7 +114,11 @@ var kubeKubeadmJoin = &cobra.Command{
}
}
cfg := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
cfg, err := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
if err != nil {
log.Error("unable to create manifest", "err", err)
return
}
fmt.Println(cfg) // output manifest to stdout
},
}

18
cmd/kube-vip-manifests.go
View File

@@ -54,7 +54,7 @@ var kubeManifestPod = &cobra.Command{
// The control plane has a requirement for a VIP being specified
if initConfig.EnableControlPlane && (initConfig.VIP == "" && initConfig.Address == "" && !initConfig.DDNS) {
_ = cmd.Help()
log.Error("No address is specified for kube-vip to expose services on")
log.Error("no address is specified for kube-vip to expose services on")
return
}
@@ -67,7 +67,11 @@ var kubeManifestPod = &cobra.Command{
}
}
cfg := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
cfg, err := kubevip.GeneratePodManifestFromConfig(&initConfig, image, Release.Version, inCluster)
if err != nil {
log.Error("unable to create manifest", "err", err)
return
}
fmt.Println(cfg) // output manifest to stdout
},
}
@@ -87,7 +91,7 @@ var kubeManifestDaemon = &cobra.Command{
// The control plane has a requirement for a VIP being specified
if initConfig.EnableControlPlane && (initConfig.VIP == "" && initConfig.Address == "" && !initConfig.DDNS) {
_ = cmd.Help()
log.Error("No address is specified for kube-vip to expose services on")
log.Error("no address is specified for kube-vip to expose services on")
return
}
@@ -100,7 +104,11 @@ var kubeManifestDaemon = &cobra.Command{
}
}
cfg := kubevip.GenerateDaemonsetManifestFromConfig(&initConfig, image, Release.Version, inCluster, taint)
cfg, err := kubevip.GenerateDaemonsetManifestFromConfig(&initConfig, image, Release.Version, inCluster, taint)
if err != nil {
log.Error("unable to create manifest", "err", err)
return
}
fmt.Println(cfg) // output manifest to stdout
},
}
@@ -121,7 +129,7 @@ var kubeManifestRbac = &cobra.Command{
// The control plane has a requirement for a VIP being specified
if initConfig.EnableControlPlane && (initConfig.VIP == "" && initConfig.Address == "" && !initConfig.DDNS) {
_ = cmd.Help()
log.Error("No address is specified for kube-vip to expose services on")
log.Error("no address is specified for kube-vip to expose services on")
return
}

31
pkg/cluster/clusterLeaderElection.go
View File

@@ -2,6 +2,7 @@ package cluster
import (
"context"
"errors"
"fmt"
"os"
"os/signal"
@@ -117,6 +118,11 @@ func (cluster *Cluster) StartCluster(ctx context.Context, c *kubevip.Config, sm
// Add Notification for SIGTERM (sent from Kubernetes)
signal.Notify(signalChan, syscall.SIGTERM)
if cluster.completed == nil {
cluster.completed = make(chan bool, 1)
defer close(cluster.completed)
}
go func() {
<-signalChan
log.Info("Received termination, signaling cluster shutdown")
@@ -184,6 +190,7 @@ func (cluster *Cluster) StartCluster(ctx context.Context, c *kubevip.Config, sm
err := cluster.vipService(clusterCtx, c, sm, bgpServer, leaderCancel)
if err != nil {
log.Error("starting VIP service on leader", "err", err)
signalChan <- syscall.SIGINT
}
},
onStoppedLeading: func() {
@@ -235,7 +242,7 @@ func (cluster *Cluster) StartCluster(ctx context.Context, c *kubevip.Config, sm
}
log.Error("lost leadership, restarting kube-vip")
panic("") // TODO - we could also return here
signalChan <- syscall.SIGINT
},
onNewLeader: func(identity string) {
// we're notified when new leader elected
@@ -264,7 +271,9 @@ func (cluster *Cluster) StartCluster(ctx context.Context, c *kubevip.Config, sm
case "kubernetes", "":
cluster.runKubernetesLeaderElectionOrDie(leaderCtx, run)
case "etcd":
cluster.runEtcdLeaderElectionOrDie(leaderCtx, run)
if err := cluster.runEtcdLeaderElectionOrDie(leaderCtx, run); err != nil {
return err
}
default:
log.Info(fmt.Sprintf("LeaderElectionMode %s not supported, exiting", c.LeaderElectionType))
}
@@ -323,8 +332,8 @@ func (cluster *Cluster) runKubernetesLeaderElectionOrDie(ctx context.Context, ru
})
}
func (cluster *Cluster) runEtcdLeaderElectionOrDie(ctx context.Context, run *runConfig) {
etcd.RunElectionOrDie(ctx, &etcd.LeaderElectionConfig{
func (cluster *Cluster) runEtcdLeaderElectionOrDie(ctx context.Context, run *runConfig) error {
if err := etcd.RunElectionOrDie(ctx, &etcd.LeaderElectionConfig{
EtcdConfig: etcd.ClientConfig{Client: run.sm.EtcdClient},
Name: run.config.LeaseName,
MemberID: run.leaseID,
@@ -334,10 +343,13 @@ func (cluster *Cluster) runEtcdLeaderElectionOrDie(ctx context.Context, run *run
OnStoppedLeading: run.onStoppedLeading,
OnNewLeader: run.onNewLeader,
},
})
}); err != nil {
return fmt.Errorf("etcd leaderelection: %w", err)
}
return nil
}
func (sm *Manager) NodeWatcher(ctxArp context.Context, lb *loadbalancer.IPVSLoadBalancer, port uint16) error {
func (sm *Manager) NodeWatcher(ctx context.Context, lb *loadbalancer.IPVSLoadBalancer, port uint16) error {
// Use a restartable watcher, as this should help in the event of etcd or timeout issues
log.Info("Kube-Vip is watching nodes for control-plane labels")
@@ -345,9 +357,9 @@ func (sm *Manager) NodeWatcher(ctxArp context.Context, lb *loadbalancer.IPVSLoad
LabelSelector: "node-role.kubernetes.io/control-plane",
}
rw, err := watchtools.NewRetryWatcherWithContext(ctxArp, "1", &cache.ListWatch{
rw, err := watchtools.NewRetryWatcherWithContext(ctx, "1", &cache.ListWatch{
WatchFunc: func(_ metav1.ListOptions) (watch.Interface, error) {
return sm.RetryWatcherClient.CoreV1().Nodes().Watch(ctxArp, listOptions)
return sm.RetryWatcherClient.CoreV1().Nodes().Watch(ctx, listOptions)
},
})
if err != nil {
@@ -379,6 +391,9 @@ func (sm *Manager) NodeWatcher(ctxArp context.Context, lb *loadbalancer.IPVSLoad
err = lb.AddBackend(node.Status.Addresses[x].Address, port)
if err != nil {
log.Error("add IPVS backend", "err", err)
if errors.Is(err, &utils.PanicError{}) {
return fmt.Errorf("add IPVS backend: %w", err)
}
}
} else {
err = lb.RemoveBackend(node.Status.Addresses[x].Address, port)

36
pkg/cluster/service.go
View File

@@ -42,6 +42,20 @@ func (cluster *Cluster) vipService(ctx context.Context, c *kubevip.Config, sm *M
// Add Notification for SIGTERM (sent from Kubernetes)
signal.Notify(signalChan, syscall.SIGTERM)
shouldClose := false
if cluster.completed == nil {
cluster.completed = make(chan bool, 1)
shouldClose = true
}
go func() {
<-ctx.Done()
signalChan <- syscall.SIGINT
if shouldClose {
defer close(cluster.completed)
}
}()
loadbalancers := []*loadbalancer.IPVSLoadBalancer{}
var arpWG sync.WaitGroup
@@ -56,8 +70,7 @@ func (cluster *Cluster) vipService(ctx context.Context, c *kubevip.Config, sm *M
}
if err := network.SetMask(c.VIPSubnet); err != nil {
log.Error("failed to set mask", "subnet", c.VIPSubnet, "err", err)
panic("")
return fmt.Errorf("failed to set mask for subnet %q: %w", c.VIPSubnet, err)
}
// start the dns updater if address is dns
@@ -86,13 +99,16 @@ func (cluster *Cluster) vipService(ctx context.Context, c *kubevip.Config, sm *M
if c.EnableLoadBalancer {
lb, err := loadbalancer.NewIPVSLB(network.IP(), c.LoadBalancerPort, c.LoadBalancerForwardingMethod, c.BackendHealthCheckInterval, c.Interface, cancelLeaderElection, signalChan)
if err != nil {
log.Error("Error creating IPVS LoadBalancer", "err", err)
return fmt.Errorf("creating IPVS LoadBalance: %w", err)
}
go func() {
err = sm.NodeWatcher(ctx, lb, c.Port) //TODO: We're using the ctxARP as the context this will change when rkatz finishes his change
err = sm.NodeWatcher(ctx, lb, c.Port)
if err != nil {
log.Error("Error watching node labels", "err", err)
if errors.Is(err, &utils.PanicError{}) {
signalChan <- syscall.SIGINT
}
}
}()
@@ -131,7 +147,7 @@ func (cluster *Cluster) vipService(ctx context.Context, c *kubevip.Config, sm *M
ips := []string{}
if nodename != "" {
if ips, err = getNodeIPs(ctx, nodename, sm.KubernetesClient); err != nil && !apierrors.IsNotFound(err) {
log.Error("failed to get IP of control-plane nod", "err", err)
log.Error("failed to get IP of control-plane node", "err", err)
}
}
@@ -231,7 +247,8 @@ func (cluster *Cluster) vipService(ctx context.Context, c *kubevip.Config, sm *M
deleted, err := network.DeleteIP()
if err != nil {
log.Error("error deleting IP", "err", err)
panic("")
signalChan <- syscall.SIGINT
return
}
if deleted {
log.Info("deleted address", "IP", network.IP(), "interface", network.Interface())
@@ -267,7 +284,7 @@ func getNodeIPs(ctx context.Context, nodename string, client *kubernetes.Clients
}
// StartLoadBalancerService will start a VIP instance and leave it for kube-proxy to handle
func (cluster *Cluster) StartLoadBalancerService(ctx context.Context, c *kubevip.Config, bgp *bgp.Server, name string, CountRouteReferences func(*netlink.Route) int) {
func (cluster *Cluster) StartLoadBalancerService(ctx context.Context, c *kubevip.Config, bgp *bgp.Server, name string, CountRouteReferences func(*netlink.Route) int) error {
// use a Go context so we can tell the arp loop code when we
// want to step down
//nolint
@@ -303,7 +320,8 @@ func (cluster *Cluster) StartLoadBalancerService(ctx context.Context, c *kubevip
log.Debug("current ip to process", "ip", network.IP(), "mask", c.VIPSubnet)
if err := network.SetMask(c.VIPSubnet); err != nil {
log.Error("failed to set mask", "subnet", c.VIPSubnet, "err", err)
panic("")
cancelArp()
return utils.NewPanicError(fmt.Sprintf("failed to set mask for subnet %q: %s", c.VIPSubnet, err.Error()))
}
_, err := network.DeleteIP()
if err != nil {
@@ -417,6 +435,8 @@ func (cluster *Cluster) StartLoadBalancerService(ctx context.Context, c *kubevip
close(cluster.completed)
}()
return nil
}
// Layer2Update, handles the creation of the

6
pkg/etcd/election.go
View File

@@ -2,6 +2,7 @@ package etcd
import (
"context"
"fmt"
"hash/fnv"
"time"
@@ -64,10 +65,11 @@ type ClientConfig struct {
}
// RunElectionOrDie behaves the same way as RunElection but panics if there is an error.
func RunElectionOrDie(ctx context.Context, config *LeaderElectionConfig) {
func RunElectionOrDie(ctx context.Context, config *LeaderElectionConfig) error {
if err := RunElection(ctx, config); err != nil {
panic(err)
return fmt.Errorf("leaderelection error: %w", err)
}
return nil
}
// RunElection starts a client with the provided config or panics.

2
pkg/kubevip/config_environment.go
View File

@@ -26,7 +26,7 @@ func ParseEnvironment(c *Config) error {
if env != "" {
logLevel, err := strconv.ParseInt(env, 10, 32)
if err != nil {
panic("Unable to parse environment variable [vip_loglevel], should be int")
return fmt.Errorf("unable to parse environment variable [vip_loglevel], should be int")
}
c.Logging = int32(logLevel)
}

32
pkg/kubevip/config_generator.go
View File

@@ -167,10 +167,10 @@ func GenerateRoleBinding(rolebinding bool, saCfg *applyCoreV1.ServiceAccountAppl
}
// generatePodSpec will take a kube-vip config and generate a Pod spec
func generatePodSpec(c *Config, image, imageVersion string, inCluster bool) *corev1.Pod {
func generatePodSpec(c *Config, image, imageVersion string, inCluster bool) (*corev1.Pod, error) {
imageRef, err := name.NewTag(image, name.WeakValidation, name.WithDefaultTag(imageVersion))
if err != nil {
panic(fmt.Errorf("cannot parse %q: %w", image, err))
return nil, fmt.Errorf("cannot parse %q: %w", image, err)
}
command := "manager"
@@ -669,18 +669,24 @@ func generatePodSpec(c *Config, image, imageVersion string, inCluster bool) *cor
newManifest.Spec.HostAliases = append(newManifest.Spec.HostAliases, hostAlias)
}
return newManifest
return newManifest, nil
}
// GeneratePodManifestFromConfig will take a kube-vip config and generate a manifest
func GeneratePodManifestFromConfig(c *Config, image, imageVersion string, inCluster bool) string {
newManifest := generatePodSpec(c, image, imageVersion, inCluster)
b, _ := yaml.Marshal(newManifest)
return string(b)
func GeneratePodManifestFromConfig(c *Config, image, imageVersion string, inCluster bool) (string, error) {
newManifest, err := generatePodSpec(c, image, imageVersion, inCluster)
if err != nil {
return "", err
}
b, err := yaml.Marshal(newManifest)
if err != nil {
return "", fmt.Errorf("failed to marshal manifest: %w", err)
}
return string(b), nil
}
// GenerateDaemonsetManifestFromConfig will take a kube-vip config and generate a manifest
func GenerateDaemonsetManifestFromConfig(c *Config, image, imageVersion string, inCluster, taint bool) string {
func GenerateDaemonsetManifestFromConfig(c *Config, image, imageVersion string, inCluster, taint bool) (string, error) {
// Determine where the pod should be deployed
var namespace string
if c.ServiceNamespace != "" {
@@ -689,7 +695,11 @@ func GenerateDaemonsetManifestFromConfig(c *Config, image, imageVersion string,
namespace = metav1.NamespaceSystem
}
podSpec := generatePodSpec(c, image, imageVersion, inCluster).Spec
pod, err := generatePodSpec(c, image, imageVersion, inCluster)
if err != nil {
return "", err
}
newManifest := &appv1.DaemonSet{
TypeMeta: metav1.TypeMeta{
Kind: "DaemonSet",
@@ -716,7 +726,7 @@ func GenerateDaemonsetManifestFromConfig(c *Config, image, imageVersion string,
"app.kubernetes.io/version": imageVersion,
},
},
Spec: podSpec,
Spec: pod.Spec,
},
},
}
@@ -766,5 +776,5 @@ func GenerateDaemonsetManifestFromConfig(c *Config, image, imageVersion string,
delete(m, "status")
b, _ = yaml.Marshal(m)
return string(b)
return string(b), nil
}

28
pkg/loadbalancer/ipvs.go
View File

@@ -66,28 +66,34 @@ func NewIPVSLB(address string, port uint16, forwardingMethod string, backendHeal
c, err := ipvs.New()
if err != nil {
log.Error("ensure IPVS kernel modules are loaded")
log.Error("Error starting IPVS", "err", err)
panic("")
log.Error("error starting IPVS", "err", err)
return nil, fmt.Errorf("starting IPVS: %w", err)
}
i, err := c.Info()
if err != nil {
log.Error("ensure IPVS kernel modules are loaded")
log.Error("Error retrieving IPVS info", "err", err)
log.Error("error retrieving IPVS info", "err", err)
if errors.Is(err, os.ErrPermission) {
log.Error("no permission to get IPVS info - please ensure that kube-vip is running with proper capabilities/privileged mode")
}
panic("")
return nil, fmt.Errorf("retrieving IPVS: %w", err)
}
log.Info("IPVS Loadbalancer enabled", "version", fmt.Sprintf("%d.%d.%d", i.Version[0], i.Version[1], i.Version[2]))
ip, family := ipAndFamily(address)
if strings.ToLower(forwardingMethod) == "masquerade" {
enableProcSys("/proc/sys/net/ipv4/vs/conntrack", "net.ipv4.vs.conntrack")
if err := enableProcSys("/proc/sys/net/ipv4/vs/conntrack", "net.ipv4.vs.conntrack"); err != nil {
return nil, err
}
if family == ipvs.INET6 {
enableProcSys("/proc/sys/net/ipv6/conf/all/forwarding", "net.ipv6.conf.all.forwarding")
if err := enableProcSys("/proc/sys/net/ipv6/conf/all/forwarding", "net.ipv6.conf.all.forwarding"); err != nil {
return nil, err
}
} else {
enableProcSys("/proc/sys/net/ipv4/ip_forward", "net.ipv4.ip_forward")
if err := enableProcSys("/proc/sys/net/ipv4/ip_forward", "net.ipv4.ip_forward"); err != nil {
return nil, err
}
}
}
@@ -148,15 +154,15 @@ func NewIPVSLB(address string, port uint16, forwardingMethod string, backendHeal
return lb, nil
}
func enableProcSys(path, name string) {
func enableProcSys(path, name string) error {
isSet, err := sysctl.EnableProcSys(path)
if err != nil {
log.Error(fmt.Sprintf("ensuring %s enabled", name), "err", err)
panic("")
return fmt.Errorf("ensuring %s enabled: %w", name, err)
}
if isSet {
log.Info(fmt.Sprintf("sysctl set %s to 1", name))
}
return nil
}
func (lb *IPVSLoadBalancer) RemoveIPVSLB() error {
@@ -228,7 +234,7 @@ func (lb *IPVSLoadBalancer) addBackend(address string, port uint16) error {
// Fatal error at this point as IPVS is probably not working
log.Error("Unable to create an IPVS service, ensure IPVS kernel modules are loaded")
log.Error("IPVS service", "err", err)
panic("")
return utils.NewPanicError(fmt.Sprintf("unable to create an IPVS service - %s", err))
}
log.Info("load-Balancer services created", "address", lb.addrString(), "port", lb.Port)

26
pkg/manager/manager.go
View File

@@ -9,6 +9,7 @@ import (
"path/filepath"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@@ -16,6 +17,7 @@ import (
"github.com/kube-vip/kube-vip/pkg/arp"
"github.com/kube-vip/kube-vip/pkg/bgp"
"github.com/kube-vip/kube-vip/pkg/cluster"
"github.com/kube-vip/kube-vip/pkg/k8s"
"github.com/kube-vip/kube-vip/pkg/kubevip"
"github.com/kube-vip/kube-vip/pkg/networkinterface"
@@ -72,6 +74,9 @@ type Manager struct {
// implementation will be decided in constructor
// based on config.EnableNodeLabeling
nodeLabelManager node.LabelManager
// This variable reports if manager is being closed
closing atomic.Bool
}
// New will create a new managing object
@@ -355,3 +360,24 @@ func (sm *Manager) parseAnnotations(ctx context.Context) error {
}
return nil
}
func (sm *Manager) waitForShutdown(ctx context.Context, cancel context.CancelFunc, cpCluster *cluster.Cluster) {
defer close(sm.shutdownChan)
for {
sig := <-sm.signalChan
switch sig {
case syscall.SIGUSR1:
log.Info("Received SIGUSR1, dumping configuration")
sm.dumpConfiguration(ctx)
case syscall.SIGINT, syscall.SIGTERM:
sm.closing.Store(true)
log.Info("Received kube-vip termination, signaling shutdown")
if cpCluster != nil {
cpCluster.Stop()
}
// Cancel the context, which will in turn cancel the leadership and all goroutines
cancel()
return
}
}
}

195
pkg/manager/manager_arp.go
View File

@@ -19,7 +19,6 @@ import (
// Start will begin the Manager, which will start services and watch the configmap
func (sm *Manager) startARP(ctx context.Context, id string) error {
var cpCluster *cluster.Cluster
var ns string
var err error
// use a Go context so we can tell the leaderelection code when we
@@ -31,26 +30,7 @@ func (sm *Manager) startARP(ctx context.Context, id string) error {
go sm.arpMgr.StartAdvertisement(arpCtx)
// Shutdown function that will wait on this signal, unless we call it ourselves
go func() {
for {
sig := <-sm.signalChan
switch sig {
case syscall.SIGUSR1:
log.Info("Received SIGUSR1, dumping configuration")
sm.dumpConfiguration(arpCtx)
case syscall.SIGINT, syscall.SIGTERM:
log.Info("Received kube-vip termination, signaling shutdown")
if sm.config.EnableControlPlane {
cpCluster.Stop()
}
// Close all go routines
close(sm.shutdownChan)
// Cancel the context, which will in turn cancel the leadership
cancel()
return
}
}
}()
go sm.waitForShutdown(arpCtx, cancel, cpCluster)
if sm.config.EnableControlPlane {
cpCluster, err = cluster.InitCluster(sm.config, false, sm.intfMgr, sm.arpMgr)
@@ -67,102 +47,103 @@ func (sm *Manager) startARP(ctx context.Context, id string) error {
err := cpCluster.StartCluster(arpCtx, sm.config, clusterManager, nil)
if err != nil {
log.Error("starting control plane", "err", err)
// Trigger the shutdown of this manager instance
}
// Trigger the shutdown of this manager instance
if !sm.closing.Load() {
sm.signalChan <- syscall.SIGINT
}
}()
}
// Check if we're also starting the services, if not we can sit and wait on the closing channel and return here
if !sm.config.EnableServices {
<-sm.shutdownChan
log.Info("Shutting down Kube-Vip")
return nil
if sm.config.EnableServices {
// This will tidy any dangling kube-vip iptables rules
if sm.config.EgressClean {
vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv4)
}
ns = sm.config.Namespace
} else {
// Start a services watcher (all kube-vip pods will watch services), upon a new service
// a lock based upon that service is created that they will all leaderElection on
if sm.config.EnableServicesElection {
log.Info("beginning watching services, leaderelection will happen for every service")
err = sm.svcProcessor.StartServicesWatchForLeaderElection(arpCtx)
if err != nil {
return err
}
} else {
ns, err := returnNameSpace()
if err != nil {
log.Warn("unable to auto-detect namespace, dropping to config", "namespace", sm.config.Namespace)
ns = sm.config.Namespace
}
ns, err = returnNameSpace()
if err != nil {
log.Warn("unable to auto-detect namespace, dropping to config", "namespace", sm.config.Namespace)
ns = sm.config.Namespace
log.Info("beginning services leadership", "namespace", ns, "lock name", sm.config.ServicesLeaseName, "id", id)
// 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{
LeaseMeta: metav1.ObjectMeta{
Name: sm.config.ServicesLeaseName,
Namespace: ns,
},
Client: sm.clientSet.CoordinationV1(),
LockConfig: resourcelock.ResourceLockConfig{
Identity: id,
},
}
// start the leader election code loop
leaderelection.RunOrDie(arpCtx, leaderelection.LeaderElectionConfig{
Lock: lock,
// IMPORTANT: you MUST ensure that any code you have that
// is protected by the lease must terminate **before**
// you call cancel. Otherwise, you could have a background
// loop still running and another process could
// get elected before your background loop finished, violating
// the stated goal of the lease.
ReleaseOnCancel: true,
LeaseDuration: time.Duration(sm.config.LeaseDuration) * time.Second,
RenewDeadline: time.Duration(sm.config.RenewDeadline) * time.Second,
RetryPeriod: time.Duration(sm.config.RetryPeriod) * time.Second,
Callbacks: leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
err = sm.svcProcessor.ServicesWatcher(ctx, sm.svcProcessor.SyncServices)
if err != nil {
log.Error("service watcher", "err", err)
if !sm.closing.Load() {
sm.signalChan <- syscall.SIGINT
}
}
},
OnStoppedLeading: func() {
// we can do cleanup here
sm.mutex.Lock()
defer sm.mutex.Unlock()
log.Info("leader lost", "new leader", id)
sm.svcProcessor.Stop()
log.Error("lost services leadership, restarting kube-vip")
if !sm.closing.Load() {
sm.signalChan <- syscall.SIGINT
}
},
OnNewLeader: func(identity string) {
// we're notified when new leader elected
if sm.config.EnableNodeLabeling {
applyNodeLabel(arpCtx, sm.clientSet, sm.config.Address, id, identity)
}
if identity == id {
// I just got the lock
return
}
log.Info("new leader elected", "new leader", identity)
},
},
})
}
}
// This will tidy any dangling kube-vip iptables rules
if sm.config.EgressClean {
vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv4)
}
<-sm.shutdownChan
log.Info("Shutting down Kube-Vip")
// Start a services watcher (all kube-vip pods will watch services), upon a new service
// a lock based upon that service is created that they will all leaderElection on
if sm.config.EnableServicesElection {
log.Info("beginning watching services, leaderelection will happen for every service")
err = sm.svcProcessor.StartServicesWatchForLeaderElection(arpCtx)
if err != nil {
return err
}
} else {
log.Info("beginning services leadership", "namespace", ns, "lock name", sm.config.ServicesLeaseName, "id", id)
// 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{
LeaseMeta: metav1.ObjectMeta{
Name: sm.config.ServicesLeaseName,
Namespace: ns,
},
Client: sm.clientSet.CoordinationV1(),
LockConfig: resourcelock.ResourceLockConfig{
Identity: id,
},
}
// start the leader election code loop
leaderelection.RunOrDie(arpCtx, leaderelection.LeaderElectionConfig{
Lock: lock,
// IMPORTANT: you MUST ensure that any code you have that
// is protected by the lease must terminate **before**
// you call cancel. Otherwise, you could have a background
// loop still running and another process could
// get elected before your background loop finished, violating
// the stated goal of the lease.
ReleaseOnCancel: true,
LeaseDuration: time.Duration(sm.config.LeaseDuration) * time.Second,
RenewDeadline: time.Duration(sm.config.RenewDeadline) * time.Second,
RetryPeriod: time.Duration(sm.config.RetryPeriod) * time.Second,
Callbacks: leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
err = sm.svcProcessor.ServicesWatcher(ctx, sm.svcProcessor.SyncServices)
if err != nil {
log.Error("service watcher", "err", err)
panic("") // TODO: - emulating log.fatal here
}
},
OnStoppedLeading: func() {
// we can do cleanup here
sm.mutex.Lock()
defer sm.mutex.Unlock()
log.Info("leader lost", "new leader", id)
sm.svcProcessor.Stop()
log.Error("lost leadership, restarting kube-vip")
panic("") // TODO: - emulating log.fatal here
},
OnNewLeader: func(identity string) {
// we're notified when new leader elected
if sm.config.EnableNodeLabeling {
applyNodeLabel(arpCtx, sm.clientSet, sm.config.Address, id, identity)
}
if identity == id {
// I just got the lock
return
}
log.Info("new leader elected", "new leader", identity)
},
},
})
}
return nil
}

60
pkg/manager/manager_bgp.go
View File

@@ -60,26 +60,7 @@ func (sm *Manager) startBGP(ctx context.Context) error {
}()
// Shutdown function that will wait on this signal, unless we call it ourselves
go func() {
for {
sig := <-sm.signalChan
switch sig {
case syscall.SIGUSR1:
log.Info("Received SIGUSR1, dumping configuration")
sm.dumpConfiguration(bgpCtx)
case syscall.SIGINT, syscall.SIGTERM:
log.Info("Received termination, signaling shutdown")
if sm.config.EnableControlPlane {
if cpCluster != nil {
cpCluster.Stop()
}
}
// Cancel the context, which will in turn cancel the leadership
cancel()
return
}
}
}()
go sm.waitForShutdown(bgpCtx, cancel, cpCluster)
if sm.config.EnableControlPlane {
cpCluster, err = cluster.InitCluster(sm.config, false, sm.intfMgr, sm.arpMgr)
@@ -101,33 +82,30 @@ func (sm *Manager) startBGP(ctx context.Context) error {
if err != nil {
log.Error("Control Plane", "err", err)
// Trigger the shutdown of this manager instance
sm.signalChan <- syscall.SIGINT
if !sm.closing.Load() {
sm.signalChan <- syscall.SIGINT
}
}
}()
}
// Check if we're also starting the services, if not we can sit and wait on the closing channel and return here
if !sm.config.EnableServices {
<-sm.signalChan
log.Info("Shutting down Kube-Vip")
return nil
}
}
if sm.config.EnableServicesElection {
log.Info("beginning watching services, leaderelection will happen for every service")
err = sm.svcProcessor.StartServicesWatchForLeaderElection(bgpCtx)
if err != nil {
return err
}
} else {
log.Info("beginning watching services without leader election")
err = sm.svcProcessor.ServicesWatcher(bgpCtx, sm.svcProcessor.SyncServices)
if err != nil {
return err
if sm.config.EnableServices {
if sm.config.EnableServicesElection {
log.Info("beginning watching services, leaderelection will happen for every service")
err = sm.svcProcessor.StartServicesWatchForLeaderElection(bgpCtx)
if err != nil {
return err
}
} else {
log.Info("beginning watching services without leader election")
err = sm.svcProcessor.ServicesWatcher(bgpCtx, sm.svcProcessor.SyncServices)
if err != nil {
return err
}
}
}
<-sm.shutdownChan
log.Info("Shutting down Kube-Vip")
return nil

77
pkg/manager/manager_table.go
View File

@@ -46,24 +46,33 @@ func (sm *Manager) startTableMode(ctx context.Context, id string) error {
}
// Shutdown function that will wait on this signal, unless we call it ourselves
go func() {
for {
sig := <-sm.signalChan
switch sig {
case syscall.SIGUSR1:
log.Info("Received SIGUSR1, dumping configuration")
sm.dumpConfiguration(rtCtx)
case syscall.SIGINT, syscall.SIGTERM:
log.Info("Received kube-vip termination, signaling shutdown")
if sm.config.EnableControlPlane {
cpCluster.Stop()
}
// Cancel the context, which will in turn cancel the leadership
cancel()
return
}
go sm.waitForShutdown(rtCtx, cancel, cpCluster)
if sm.config.EnableControlPlane {
log.Debug("initCluster for ControlPlane")
cpCluster, err = cluster.InitCluster(sm.config, false, sm.intfMgr, sm.arpMgr)
if err != nil {
log.Debug("init of ControlPlane NOT successful")
return fmt.Errorf("cluster initialization error: %w", err)
}
}()
log.Debug("init of ControlPlane successful")
log.Debug("init ClusterManager")
clusterManager, err := initClusterManager(sm)
if err != nil {
log.Debug("init cluster manager NOT successful")
return fmt.Errorf("cluster manager initialization error: %w", err)
}
log.Debug("init ClusterManager successful")
if err := cpCluster.StartVipService(rtCtx, sm.config, clusterManager, nil); err != nil {
log.Error("Control Plane", "err", err)
// Trigger the shutdown of this manager instance
if !sm.closing.Load() {
sm.signalChan <- syscall.SIGINT
}
} else {
log.Debug("start VipServer for cluster manager successful")
}
}
if sm.config.EnableServices {
log.Debug("starting Services")
@@ -114,7 +123,7 @@ func (sm *Manager) startTableMode(ctx context.Context, id string) error {
err = sm.svcProcessor.ServicesWatcher(ctx, sm.svcProcessor.SyncServices)
if err != nil {
log.Error(err.Error())
panic("")
sm.signalChan <- syscall.SIGINT
}
},
OnStoppedLeading: func() {
@@ -124,8 +133,8 @@ func (sm *Manager) startTableMode(ctx context.Context, id string) error {
log.Info("leader lost", "id", id)
sm.svcProcessor.Stop()
log.Error("lost leadership, restarting kube-vip")
panic("")
log.Error("lost services leadership, restarting kube-vip")
sm.signalChan <- syscall.SIGINT
},
OnNewLeader: func(identity string) {
// we're notified when new leader elected
@@ -141,35 +150,15 @@ func (sm *Manager) startTableMode(ctx context.Context, id string) error {
log.Info("beginning watching services without leader election")
err = sm.svcProcessor.ServicesWatcher(rtCtx, sm.svcProcessor.SyncServices)
if err != nil {
log.Error("Cannot watch services", "err", err)
return fmt.Errorf("cannot watch services: %w", err)
} else {
log.Debug("watching services")
}
}
}
if sm.config.EnableControlPlane {
log.Debug("initCluster for ControlPlane")
cpCluster, err = cluster.InitCluster(sm.config, false, sm.intfMgr, sm.arpMgr)
if err != nil {
log.Debug("init of ControlPlane NOT successful")
return fmt.Errorf("cluster initialization error: %w", err)
}
log.Debug("init of ControlPlane successful")
log.Debug("init ClusterManager")
clusterManager, err := initClusterManager(sm)
if err != nil {
log.Debug("init cluster manager NOT successful")
return fmt.Errorf("cluster manager initialization error: %w", err)
}
log.Debug("init ClusterManager successful")
if err := cpCluster.StartVipService(rtCtx, sm.config, clusterManager, nil); err != nil {
log.Error("Control Plane", "err", err)
// Trigger the shutdown of this manager instance
sm.signalChan <- syscall.SIGINT
} else {
log.Debug("start VipServer for cluster manager successful")
}
}
<-sm.shutdownChan
log.Info("Shutting down Kube-Vip")
return nil
}

34
pkg/manager/manager_wireguard.go
View File

@@ -2,6 +2,7 @@ package manager
import (
"context"
"sync/atomic"
"syscall"
"time"
@@ -38,22 +39,10 @@ func (sm *Manager) startWireguard(ctx context.Context, id string) error {
return err
}
var closing atomic.Bool
// Shutdown function that will wait on this signal, unless we call it ourselves
go func() {
for {
sig := <-sm.signalChan
switch sig {
case syscall.SIGUSR1:
log.Info("Received SIGUSR1, dumping configuration")
sm.dumpConfiguration(wgCtx)
case syscall.SIGINT, syscall.SIGTERM:
log.Info("Received termination, signaling shutdown")
// Cancel the context, which will in turn cancel the leadership
cancel()
return
}
}
}()
go sm.waitForShutdown(wgCtx, cancel, nil)
ns, err = returnNameSpace()
if err != nil {
@@ -70,7 +59,6 @@ func (sm *Manager) startWireguard(ctx context.Context, id string) error {
return err
}
} else {
log.Info("beginning services leadership", "namespace", ns, "lock name", plunderLock, "id", id)
// we use the Lease lock type since edits to Leases are less common
// and fewer objects in the cluster watch "all Leases".
@@ -103,7 +91,9 @@ func (sm *Manager) startWireguard(ctx context.Context, id string) error {
err = sm.svcProcessor.ServicesWatcher(ctx, sm.svcProcessor.SyncServices)
if err != nil {
log.Error(err.Error())
panic("")
if !closing.Load() {
sm.signalChan <- syscall.SIGINT
}
}
},
OnStoppedLeading: func() {
@@ -113,8 +103,10 @@ func (sm *Manager) startWireguard(ctx context.Context, id string) error {
log.Info("leader lost", "id", id)
sm.svcProcessor.Stop()
log.Error("lost leadership, restarting kube-vip")
panic("")
log.Error("lost services leadership, restarting kube-vip")
if !closing.Load() {
sm.signalChan <- syscall.SIGINT
}
},
OnNewLeader: func(identity string) {
// we're notified when new leader elected
@@ -127,5 +119,9 @@ func (sm *Manager) startWireguard(ctx context.Context, id string) error {
},
})
}
<-sm.shutdownChan
log.Info("Shutting down Kube-Vip")
return nil
}

32
pkg/manager/watch_annotations.go
View File

@@ -57,30 +57,21 @@ func (sm *Manager) annotationsWatcher(ctx context.Context) error {
// they're as needed
log.Warn(err.Error())
// TODO, will need refactoring as part of rikatz work
rw, err := watchtools.NewRetryWatcherWithContext(ctx, node.ResourceVersion, &cache.ListWatch{
watcherCtx, cancel := context.WithCancel(ctx)
defer cancel()
rw, err := watchtools.NewRetryWatcherWithContext(watcherCtx, node.ResourceVersion, &cache.ListWatch{
WatchFunc: func(_ metav1.ListOptions) (watch.Interface, error) {
return sm.rwClientSet.CoreV1().Nodes().Watch(ctx, listOptions)
return sm.rwClientSet.CoreV1().Nodes().Watch(watcherCtx, listOptions)
},
})
if err != nil {
return fmt.Errorf("error creating annotations watcher: %s", err.Error())
}
exitFunction := make(chan struct{})
go func() {
select {
case <-sm.shutdownChan:
log.Debug("[annotations] shutdown called")
// Stop the retry watcher
rw.Stop()
return
case <-exitFunction:
log.Debug("[annotations] function ending")
// Stop the retry watcher
rw.Stop()
return
}
defer func() {
rw.Stop()
log.Debug("[annotations] watcher stopped")
}()
ch := rw.ResultChan()
@@ -103,7 +94,8 @@ func (sm *Manager) annotationsWatcher(ctx context.Context) error {
sm.config.BGPConfig = bgpConfig
sm.config.BGPPeerConfig = bgpPeer
rw.Stop()
log.Info("[annotations] exiting Annotations watcher - annotations found")
return nil
case watch.Deleted:
node, ok := event.Object.(*v1.Node)
if !ok {
@@ -130,10 +122,8 @@ func (sm *Manager) annotationsWatcher(ctx context.Context) error {
default:
}
}
close(exitFunction)
log.Info("Exiting Annotations watcher")
log.Info("[annotations] exiting annotations watcher")
return nil
}
// parseNodeAnnotations parses the annotations on the node and updates the configuration

1
pkg/services/leader.go
View File

@@ -95,6 +95,7 @@ func (p *Processor) StartServicesLeaderElection(svcCtx *servicecontext.Context,
<-svcCtx.Ctx.Done()
if svcCtx.IsActive {
// we have no context left here so we use a new one
if err := p.deleteService(context.TODO(), service.UID); err != nil {
log.Error("service deletion", "uid", service.UID, "err", err)
}

15
pkg/services/processor.go
View File

@@ -2,6 +2,7 @@ package services
import (
"context"
"errors"
"fmt"
log "log/slog"
"reflect"
@@ -16,6 +17,7 @@ import (
"github.com/kube-vip/kube-vip/pkg/lease"
"github.com/kube-vip/kube-vip/pkg/networkinterface"
"github.com/kube-vip/kube-vip/pkg/servicecontext"
"github.com/kube-vip/kube-vip/pkg/utils"
"github.com/kube-vip/kube-vip/pkg/vip"
"github.com/prometheus/client_golang/prometheus"
"github.com/vishvananda/netlink"
@@ -212,6 +214,9 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
err = serviceFunc(svcCtx, svc)
if err != nil {
log.Error(err.Error())
if errors.Is(err, &utils.PanicError{}) {
return false, err
}
}
}
@@ -243,6 +248,9 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
err = serviceFunc(svcCtx, svc)
if err != nil {
log.Error(err.Error())
if errors.Is(err, &utils.PanicError{}) {
return false, err
}
}
go func() {
@@ -278,9 +286,11 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
if !svcCtx.IsActive {
log.Info("(svcs) restartable service watcher starting", "uid", svc.UID)
err = serviceFunc(svcCtx, svc)
if err != nil {
log.Error(err.Error())
if errors.Is(err, &utils.PanicError{}) {
svcCtx.Cancel()
}
}
log.Info("(svcs) restartable service watcher done", "uid", svc.UID)
}
@@ -293,6 +303,9 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF
err = serviceFunc(svcCtx, svc)
if err != nil {
log.Error(err.Error())
if errors.Is(err, &utils.PanicError{}) {
return false, err
}
}
}
if !p.config.EnableServicesElection {

4
pkg/services/services.go
View File

@@ -163,7 +163,9 @@ func (p *Processor) addService(ctx context.Context, svc *v1.Service) error {
for x := range newService.VIPConfigs {
log.Debug("starting loadbalancer for service", "name", svc.Name, "namespace", svc.Namespace, "uid", svc.UID)
newService.Clusters[x].StartLoadBalancerService(ctx, newService.VIPConfigs[x], p.bgpServer, svc.Name, p.CountRouteReferences)
if err := newService.Clusters[x].StartLoadBalancerService(ctx, newService.VIPConfigs[x], p.bgpServer, svc.Name, p.CountRouteReferences); err != nil {
return fmt.Errorf("failed to start lb: %w", err)
}
}
p.upnpMap(ctx, newService)

15
pkg/utils/panic.go Normal file
View File

@@ -0,0 +1,15 @@
package utils
import "fmt"
type PanicError struct {
cause string
}
func (e *PanicError) Error() string {
return fmt.Sprintf("%s - unrecoverable error", e.cause)
}
func NewPanicError(cause string) error {
return &PanicError{cause: cause}
}

3
testing/e2e/e2e_bgp_test.go
View File

@@ -1046,7 +1046,7 @@ func testServiceBGP(ctx context.Context, svcName, lbAddress string, trafficPolic
func GetContainerIPs(ctx context.Context, containerName string) (string, string, error) {
cli, err := client.NewClientWithOpts(client.FromEnv)
if err != nil {
panic(err)
return "", "", fmt.Errorf("failed to create client: %w", err)
}
containers, err := cli.ContainerList(ctx, container.ListOptions{})
if err != nil {
@@ -1056,7 +1056,6 @@ func GetContainerIPs(ctx context.Context, containerName string) (string, string,
for _, c := range containers {
for _, n := range c.Names {
if n[1:] == containerName {
fmt.Println(n)
for _, n := range c.NetworkSettings.Networks {
return n.IPAddress, n.GlobalIPv6Address, nil
}

9
testing/e2e/e2e_test.go
View File

@@ -899,7 +899,7 @@ var _ = Describe("kube-vip ARP/NDP broadcast neighbor", Ordered, func() {
}
})
func createKindCluster(logger log.Logger, config *v1alpha4.Cluster, clusterName string) (kubernetes.Interface, *rest.Config) {
func createKindCluster(logger log.Logger, config *v1alpha4.Cluster, clusterName string) (kubernetes.Interface, *rest.Config, error) {
provider := cluster.NewProvider(
cluster.ProviderWithLogger(logger),
cluster.ProviderWithDocker(),
@@ -920,13 +920,13 @@ func createKindCluster(logger log.Logger, config *v1alpha4.Cluster, clusterName
cfg, err := clientcmd.BuildConfigFromKubeconfigGetter("", kubeconfigGetter)
if err != nil {
panic(err.Error())
return nil, nil, fmt.Errorf("failed to build kubeconfig: %w", err)
}
client, err := kubernetes.NewForConfig(cfg)
Expect(err).ToNot(HaveOccurred())
return client, cfg
return client, cfg, nil
}
// Assume the VIP is routable if status code is 200 or 500. Since etcd might glitch.
@@ -1231,7 +1231,8 @@ func prepareCluster(ctx context.Context, tempDirPath, clusterNameSuffix, k8sImag
clusterName := fmt.Sprintf("%s-%s", filepath.Base(tempDirPath), clusterNameSuffix)
By(withTimestamp("creating a kind cluster with multiple control plane nodes"))
client, cfg := createKindCluster(logger, &clusterConfig, clusterName)
client, cfg, err := createKindCluster(logger, &clusterConfig, clusterName)
Expect(err).ToNot(HaveOccurred())
By(withTimestamp("creating test daemonset"))
for i := range dsNumber {