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

This branch contains the code for leader election

Browse Source 
This branch adds the option for using the Kubernetes leader election algorithim to ensure that one node in the cluster will host the virtual IP and load-balancer. To use leaderElection the use teh `--leaderElection` flag when running `kubeadm init`
This commit is contained in:
thebsdbox
2020-07-03 14:02:55 +01:00
parent a9434382d1
commit 147f130e6e
8 changed files with 625 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@@ -6,7 +6,7 @@ TARGET := kube-vip
.DEFAULT_GOAL: $(TARGET)
# These will be provided to the target
VERSION := 0.1.5
VERSION := 0.1.6-election
BUILD := `git rev-parse HEAD`
# Operating System Default (LINUX)

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

@@ -37,6 +37,7 @@ func init() {
kubeKubeadm.PersistentFlags().BoolVar(&initConfig.AddPeersAsBackends, "addPeersToLB", true, "The Virtual IP addres")
kubeKubeadm.PersistentFlags().BoolVar(&initConfig.GratuitousARP, "arp", true, "Enable Arp for Vip changes")
kubeKubeadm.PersistentFlags().BoolVar(&initConfig.EnableLeaderElection, "leaderElection", false, "Use the Kubernetes leader election mechanism for clustering")
// Load Balancer flags
kubeKubeadm.PersistentFlags().BoolVar(&initLoadBalancer.BindToVip, "lbBindToVip", true, "Bind example load balancer to VIP")

45
cmd/kube-vip-start.go
View File

@@ -13,8 +13,9 @@ import (
// Start as a single node (no cluster), start as a leader in the cluster
var startConfig kubevip.Config
var startConfigLB kubevip.LoadBalancer
var startLocalPeer string
var startLocalPeer, startKubeConfigPath string
var startRemotePeers, startBackends []string
var inCluster bool
func init() {
// Get the configuration file
@@ -36,6 +37,12 @@ func init() {
kubeVipStart.Flags().IntVar(&startConfigLB.Port, "lbPort", 8080, "Port that load balander will expose on")
kubeVipStart.Flags().IntVar(&startConfigLB.BackendPort, "lbBackEndPort", 6443, "A port that all backends may be using (optional)")
kubeVipStart.Flags().StringSliceVar(&startBackends, "lbBackends", []string{"192.168.0.1:8080", "192.168.0.2:8080"}, "Comma seperated backends, format: address:port")
// Cluster configuration
kubeVipStart.Flags().StringVar(&startKubeConfigPath, "kubeConfig", "/etc/kubernetes/admin.conf", "The path of a kubernetes configuration file")
kubeVipStart.Flags().BoolVar(&inCluster, "inCluster", false, "Use the incluster token to authenticate to Kubernetes")
kubeVipStart.Flags().BoolVar(&startConfig.EnableLeaderElection, "leaderElection", false, "Use the Kubernetes leader election mechanism for clustering")
}
var kubeVipStart = &cobra.Command{
@@ -83,19 +90,33 @@ var kubeVipStart = &cobra.Command{
log.Fatalf("%v", err)
}
// Start a multi-node (raft) cluster
err = newCluster.StartCluster(&startConfig)
if err != nil {
log.Fatalf("%v", err)
if startConfig.EnableLeaderElection {
cm, err := cluster.NewManager(startKubeConfigPath, inCluster)
if err != nil {
log.Fatalf("%v", err)
}
// Leader Cluster will block
err = newCluster.StartLeaderCluster(&startConfig, cm)
if err != nil {
log.Fatalf("%v", err)
}
} else {
// // Start a multi-node (raft) cluster, this doesn't block so will wait on signal
err = newCluster.StartRaftCluster(&startConfig)
if err != nil {
log.Fatalf("%v", err)
}
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, os.Interrupt)
<-signalChan
newCluster.Stop()
}
}
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, os.Interrupt)
<-signalChan
newCluster.Stop()
},
}

290
pkg/cluster/cluster.go
View File

@@ -1,15 +1,8 @@
package cluster
import (
"fmt"
"net"
"time"
"github.com/hashicorp/raft"
"github.com/plunder-app/kube-vip/pkg/kubevip"
"github.com/plunder-app/kube-vip/pkg/loadbalancer"
"github.com/plunder-app/kube-vip/pkg/vip"
log "github.com/sirupsen/logrus"
)
const leaderLogcount = 5
@@ -54,286 +47,3 @@ func startNetworking(c *kubevip.Config) (*vip.Network, error) {
}
return &network, nil
}
// StartCluster - Begins a running instance of the Raft cluster
func (cluster *Cluster) StartCluster(c *kubevip.Config) error {
// Create local configuration address
localAddress := fmt.Sprintf("%s:%d", c.LocalPeer.Address, c.LocalPeer.Port)
// Begin the Raft configuration
config := raft.DefaultConfig()
config.LocalID = raft.ServerID(c.LocalPeer.ID)
logger := log.StandardLogger().Writer()
config.LogOutput = logger
// Initialize communication
address, err := net.ResolveTCPAddr("tcp", localAddress)
if err != nil {
return err
}
// Create transport
transport, err := raft.NewTCPTransport(localAddress, address, 3, 10*time.Second, logger)
if err != nil {
return err
}
// Create Raft structures
snapshots := raft.NewInmemSnapshotStore()
logStore := raft.NewInmemStore()
stableStore := raft.NewInmemStore()
// Cluster configuration
configuration := raft.Configuration{}
// Add Local Peer
configuration.Servers = append(configuration.Servers, raft.Server{
ID: raft.ServerID(c.LocalPeer.ID),
Address: raft.ServerAddress(fmt.Sprintf("%s:%d", c.LocalPeer.Address, c.LocalPeer.Port))})
// If we want to start a node as leader then we will not add any remote peers, this will leave this as a cluster of one
// The remotePeers will add themselves to the cluster as they're added
if c.StartAsLeader != true {
for x := range c.RemotePeers {
// Make sure that we don't add in this server twice
if c.LocalPeer.Address != c.RemotePeers[x].Address {
// Build the address from the peer configuration
peerAddress := fmt.Sprintf("%s:%d", c.RemotePeers[x].Address, c.RemotePeers[x].Port)
// Set this peer into the raft configuration
configuration.Servers = append(configuration.Servers, raft.Server{
ID: raft.ServerID(c.RemotePeers[x].ID),
Address: raft.ServerAddress(peerAddress)})
}
}
log.Info("This node will attempt to start as Follower")
} else {
log.Info("This node will attempt to start as Leader")
}
// Bootstrap cluster
if err := raft.BootstrapCluster(config, logStore, stableStore, snapshots, transport, configuration); err != nil {
return err
}
// Create RAFT instance
raftServer, err := raft.NewRaft(config, cluster.stateMachine, logStore, stableStore, snapshots, transport)
if err != nil {
return err
}
cluster.stop = make(chan bool, 1)
cluster.completed = make(chan bool, 1)
ticker := time.NewTicker(time.Second)
isLeader := c.StartAsLeader
// (attempt to) Remove the virtual IP, incase it already exists
cluster.network.DeleteIP()
// leader log broadcast - this counter is used to stop flooding STDOUT with leader log entries
var leaderbroadcast int
// Managers for Vip load balancers and none-vip loadbalancers
nonVipLB := loadbalancer.LBManager{}
VipLB := loadbalancer.LBManager{}
// Iterate through all Configurations
for x := range c.LoadBalancers {
// If the load balancer doesn't bind to the VIP
if c.LoadBalancers[x].BindToVip == false {
err = nonVipLB.Add("", &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
// On a cold start the node will sleep for 5 seconds to ensure that leader elections are complete
log.Infoln("This instance will wait approximately 5 seconds, from cold start to ensure cluster elections are complete")
time.Sleep(time.Second * 5)
go func() {
for {
if c.AddPeersAsBackends == true {
// Get addresses and change backends
// c.LoadBalancers[0].Backends
// for x := range raftServer.GetConfiguration().Configuration().Servers {
// raftServer.GetConfiguration().Configuration().Servers[x].Address
// }
}
// Broadcast the current leader on this node if it's the correct time (every leaderLogcount * time.Second)
if leaderbroadcast == leaderLogcount {
log.Infof("The Node [%s] is leading", raftServer.Leader())
// Reset the timer
leaderbroadcast = 0
// ensure that if this node is the leader, it is set as the leader
if localAddress == string(raftServer.Leader()) {
// Re-broadcast arp to ensure network stays up to date
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
if !isLeader {
log.Infoln("This node is leading, but isnt the leader (correcting)")
isLeader = true
}
} else {
// (attempt to) Remove the virtual IP, incase it already exists to keep nodes clean
cluster.network.DeleteIP()
isLeader = false
}
}
leaderbroadcast++
select {
case leader := <-raftServer.LeaderCh():
log.Infoln("New Election event")
if leader {
isLeader = true
log.Info("This node is assuming leadership of the cluster")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
log.Errorf("Dropping Leadership to another node in the cluster")
raftServer.LeadershipTransfer()
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
} else {
isLeader = false
log.Info("This node is becoming a follower within the cluster")
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
case <-ticker.C:
if isLeader {
result, err := cluster.network.IsSet()
if err != nil {
log.WithFields(log.Fields{"error": err, "ip": cluster.network.IP(), "interface": cluster.network.Interface()}).Error("Could not check ip")
}
if result == false {
log.Error("This node is leader and is adopting the virtual IP")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
}
}
case <-cluster.stop:
log.Info("[RAFT] Stopping this node")
log.Info("[LOADBALANCER] Stopping load balancers")
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
// Stop all load balancers associated with the Host
err = nonVipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
if isLeader {
log.Info("[VIP] Releasing the Virtual IP")
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
close(cluster.completed)
return
}
}
}()
log.Info("Started")
return nil
}
// Stop - Will stop the Cluster and release VIP if needed
func (cluster *Cluster) Stop() {
// Close the stop chanel, which will shut down the VIP (if needed)
close(cluster.stop)
// Wait until the completed channel is closed, signallign all shutdown tasks completed
<-cluster.completed
log.Info("Stopped")
}

257
pkg/cluster/clusterLeader.go Normal file
View File

@@ -0,0 +1,257 @@
package cluster
import (
"context"
"fmt"
"os"
"os/signal"
"path/filepath"
"syscall"
"time"
"github.com/plunder-app/kube-vip/pkg/kubevip"
"github.com/plunder-app/kube-vip/pkg/loadbalancer"
"github.com/plunder-app/kube-vip/pkg/vip"
log "github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/tools/leaderelection"
"k8s.io/client-go/tools/leaderelection/resourcelock"
)
const plunderLock = "plunder-lock"
const namespace = "kube-system"
// Manager degines the manager of the load-balancing services
type Manager struct {
clientSet *kubernetes.Clientset
}
// NewManager will create a new managing object
func NewManager(path string, inCluster bool) (*Manager, error) {
var clientset *kubernetes.Clientset
if inCluster {
// This will attempt to load the configuration when running within a POD
cfg, err := rest.InClusterConfig()
if err != nil {
return nil, fmt.Errorf("error creating kubernetes client config: %s", err.Error())
}
clientset, err = kubernetes.NewForConfig(cfg)
if err != nil {
return nil, fmt.Errorf("error creating kubernetes client: %s", err.Error())
}
// use the current context in kubeconfig
} else {
if path == "" {
path = filepath.Join(os.Getenv("HOME"), ".kube", "config")
}
config, err := clientcmd.BuildConfigFromFlags("", path)
if err != nil {
panic(err.Error())
}
// We modify the config so that we can always speak to the correct host
id, err := os.Hostname()
if err != nil {
return nil, err
}
// TODO - we need to make the port configurable
config.Host = fmt.Sprintf("%s:6443", id)
clientset, err = kubernetes.NewForConfig(config)
if err != nil {
return nil, fmt.Errorf("error creating kubernetes client: %s", err.Error())
}
}
return &Manager{
clientSet: clientset,
}, nil
}
// StartLeaderCluster - Begins a running instance of the Raft cluster
func (cluster *Cluster) StartLeaderCluster(c *kubevip.Config, sm *Manager) error {
id, err := os.Hostname()
if err != nil {
return err
}
log.Infof("Beginning cluster membership, namespace [%s], lock name [%s], id [%s]", namespace, plunderLock, 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: plunderLock,
Namespace: namespace,
},
Client: sm.clientSet.CoordinationV1(),
LockConfig: resourcelock.ResourceLockConfig{
Identity: id,
},
}
// use a Go context so we can tell the leaderelection code when we
// want to step down
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// listen for interrupts or the Linux SIGTERM signal and cancel
// our context, which the leader election code will observe and
// step down
signalChan := make(chan os.Signal, 1)
// Add Notification for Userland interrupt
signal.Notify(signalChan, syscall.SIGINT)
// Add Notification for SIGTERM (sent from Kubernetes)
signal.Notify(signalChan, syscall.SIGTERM)
// Add Notification for SIGKILL (sent from Kubernetes)
signal.Notify(signalChan, syscall.SIGKILL)
go func() {
<-signalChan
log.Info("Received termination, signaling shutdown")
// Cancel the context, which will in turn cancel the leadership
cancel()
}()
// (attempt to) Remove the virtual IP, incase it already exists
cluster.network.DeleteIP()
// Managers for Vip load balancers and none-vip loadbalancers
nonVipLB := loadbalancer.LBManager{}
VipLB := loadbalancer.LBManager{}
// Iterate through all Configurations
if len(c.LoadBalancers) != 0 {
for x := range c.LoadBalancers {
// If the load balancer doesn't bind to the VIP
if c.LoadBalancers[x].BindToVip == false {
err = nonVipLB.Add("", &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
}
// start the leader election code loop
leaderelection.RunOrDie(ctx, 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: 5 * time.Second,
RenewDeadline: 3 * time.Second,
RetryPeriod: 1 * time.Second,
Callbacks: leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
// we're notified when we start
log.Info("This node is assuming leadership of the cluster")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
},
OnStoppedLeading: func() {
// we can do cleanup here
log.Infof("leader lost: %s", id)
log.Info("This node is becoming a follower within the cluster")
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
},
OnNewLeader: func(identity string) {
// we're notified when new leader elected
if identity == id {
result, err := cluster.network.IsSet()
if err != nil {
log.WithFields(log.Fields{"error": err, "ip": cluster.network.IP(), "interface": cluster.network.Interface()}).Error("Could not check ip")
}
if result == false {
log.Error("This node is leader and is adopting the virtual IP")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
}
}
log.Infof("new leader elected: %s", identity)
},
},
})
//<-signalChan
log.Infof("Shutting down Kube-Vip")
return nil
}

296
pkg/cluster/clusterRaft.go Normal file
View File

@@ -0,0 +1,296 @@
package cluster
import (
"fmt"
"net"
"time"
"github.com/hashicorp/raft"
"github.com/plunder-app/kube-vip/pkg/kubevip"
"github.com/plunder-app/kube-vip/pkg/loadbalancer"
"github.com/plunder-app/kube-vip/pkg/vip"
log "github.com/sirupsen/logrus"
)
// StartRaftCluster - Begins a running instance of the Raft cluster
func (cluster *Cluster) StartRaftCluster(c *kubevip.Config) error {
// Create local configuration address
localAddress := fmt.Sprintf("%s:%d", c.LocalPeer.Address, c.LocalPeer.Port)
// Begin the Raft configuration
config := raft.DefaultConfig()
config.LocalID = raft.ServerID(c.LocalPeer.ID)
logger := log.StandardLogger().Writer()
config.LogOutput = logger
// Initialize communication
address, err := net.ResolveTCPAddr("tcp", localAddress)
if err != nil {
return err
}
// Create transport
transport, err := raft.NewTCPTransport(localAddress, address, 3, 10*time.Second, logger)
if err != nil {
return err
}
// Create Raft structures
snapshots := raft.NewInmemSnapshotStore()
logStore := raft.NewInmemStore()
stableStore := raft.NewInmemStore()
// Cluster configuration
configuration := raft.Configuration{}
// Add Local Peer
configuration.Servers = append(configuration.Servers, raft.Server{
ID: raft.ServerID(c.LocalPeer.ID),
Address: raft.ServerAddress(fmt.Sprintf("%s:%d", c.LocalPeer.Address, c.LocalPeer.Port))})
// If we want to start a node as leader then we will not add any remote peers, this will leave this as a cluster of one
// The remotePeers will add themselves to the cluster as they're added
if c.StartAsLeader != true {
for x := range c.RemotePeers {
// Make sure that we don't add in this server twice
if c.LocalPeer.Address != c.RemotePeers[x].Address {
// Build the address from the peer configuration
peerAddress := fmt.Sprintf("%s:%d", c.RemotePeers[x].Address, c.RemotePeers[x].Port)
// Set this peer into the raft configuration
configuration.Servers = append(configuration.Servers, raft.Server{
ID: raft.ServerID(c.RemotePeers[x].ID),
Address: raft.ServerAddress(peerAddress)})
}
}
log.Info("This node will attempt to start as Follower")
} else {
log.Info("This node will attempt to start as Leader")
}
// Bootstrap cluster
if err := raft.BootstrapCluster(config, logStore, stableStore, snapshots, transport, configuration); err != nil {
return err
}
// Create RAFT instance
raftServer, err := raft.NewRaft(config, cluster.stateMachine, logStore, stableStore, snapshots, transport)
if err != nil {
return err
}
cluster.stop = make(chan bool, 1)
cluster.completed = make(chan bool, 1)
ticker := time.NewTicker(time.Second)
isLeader := c.StartAsLeader
// (attempt to) Remove the virtual IP, incase it already exists
cluster.network.DeleteIP()
// leader log broadcast - this counter is used to stop flooding STDOUT with leader log entries
var leaderbroadcast int
// Managers for Vip load balancers and none-vip loadbalancers
nonVipLB := loadbalancer.LBManager{}
VipLB := loadbalancer.LBManager{}
// Iterate through all Configurations
for x := range c.LoadBalancers {
// If the load balancer doesn't bind to the VIP
if c.LoadBalancers[x].BindToVip == false {
err = nonVipLB.Add("", &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
// On a cold start the node will sleep for 5 seconds to ensure that leader elections are complete
log.Infoln("This instance will wait approximately 5 seconds, from cold start to ensure cluster elections are complete")
time.Sleep(time.Second * 5)
go func() {
for {
if c.AddPeersAsBackends == true {
// Get addresses and change backends
// c.LoadBalancers[0].Backends
// for x := range raftServer.GetConfiguration().Configuration().Servers {
// raftServer.GetConfiguration().Configuration().Servers[x].Address
// }
}
// Broadcast the current leader on this node if it's the correct time (every leaderLogcount * time.Second)
if leaderbroadcast == leaderLogcount {
log.Infof("The Node [%s] is leading", raftServer.Leader())
// Reset the timer
leaderbroadcast = 0
// ensure that if this node is the leader, it is set as the leader
if localAddress == string(raftServer.Leader()) {
// Re-broadcast arp to ensure network stays up to date
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
if !isLeader {
log.Infoln("This node is leading, but isnt the leader (correcting)")
isLeader = true
}
} else {
// (attempt to) Remove the virtual IP, incase it already exists to keep nodes clean
cluster.network.DeleteIP()
isLeader = false
}
}
leaderbroadcast++
select {
case leader := <-raftServer.LeaderCh():
log.Infoln("New Election event")
if leader {
isLeader = true
log.Info("This node is assuming leadership of the cluster")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
log.Errorf("Dropping Leadership to another node in the cluster")
raftServer.LeadershipTransfer()
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
} else {
isLeader = false
log.Info("This node is becoming a follower within the cluster")
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
case <-ticker.C:
if isLeader {
result, err := cluster.network.IsSet()
if err != nil {
log.WithFields(log.Fields{"error": err, "ip": cluster.network.IP(), "interface": cluster.network.Interface()}).Error("Could not check ip")
}
if result == false {
log.Error("This node is leader and is adopting the virtual IP")
err = cluster.network.AddIP()
if err != nil {
log.Warnf("%v", err)
}
// Once we have the VIP running, start the load balancer(s) that bind to the VIP
for x := range c.LoadBalancers {
if c.LoadBalancers[x].BindToVip == true {
err = VipLB.Add(c.VIP, &c.LoadBalancers[x])
if err != nil {
log.Warnf("Error creating loadbalancer [%s] type [%s] -> error [%s]", c.LoadBalancers[x].Name, c.LoadBalancers[x].Type, err)
}
}
}
if c.GratuitousARP == true {
// Gratuitous ARP, will broadcast to new MAC <-> IP
err = vip.ARPSendGratuitous(c.VIP, c.Interface)
if err != nil {
log.Warnf("%v", err)
}
}
}
}
case <-cluster.stop:
log.Info("[RAFT] Stopping this node")
log.Info("[LOADBALANCER] Stopping load balancers")
// Stop all load balancers associated with the VIP
err = VipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
// Stop all load balancers associated with the Host
err = nonVipLB.StopAll()
if err != nil {
log.Warnf("%v", err)
}
if isLeader {
log.Info("[VIP] Releasing the Virtual IP")
err = cluster.network.DeleteIP()
if err != nil {
log.Warnf("%v", err)
}
}
close(cluster.completed)
return
}
}
}()
log.Info("Started")
return nil
}
// Stop - Will stop the Cluster and release VIP if needed
func (cluster *Cluster) Stop() {
// Close the stop chanel, which will shut down the VIP (if needed)
close(cluster.stop)
// Wait until the completed channel is closed, signallign all shutdown tasks completed
<-cluster.completed
log.Info("Stopped")
}

34
pkg/kubevip/config_generator.go
View File

@@ -17,6 +17,9 @@ const (
//vipArp - defines if the arp broadcast should be enabled
vipArp = "vip_arp"
//vipLeaderElection - defines if the kubernetes algorithim should be used
vipLeaderElection = "vip_leaderelection"
//vipLogLevel - defines the level of logging to produce (5 being the most verbose)
vipLogLevel = "vip_loglevel"
@@ -75,6 +78,16 @@ func ParseEnvironment(c *Config) error {
c.Interface = env
}
// Find Single Node
env = os.Getenv(vipLeaderElection)
if env != "" {
b, err := strconv.ParseBool(env)
if err != nil {
return err
}
c.EnableLeaderElection = b
}
// Find vip address
env = os.Getenv(vipAddress)
if env != "" {
@@ -247,6 +260,10 @@ func GenerateManifestFromConfig(c *Config, imageVersion string) string {
Name: vipArp,
Value: strconv.FormatBool(c.GratuitousARP),
},
{
Name: vipLeaderElection,
Value: strconv.FormatBool(c.EnableLeaderElection),
},
{
Name: vipInterface,
Value: c.Interface,
@@ -333,11 +350,28 @@ func GenerateManifestFromConfig(c *Config, imageVersion string) string {
"start",
},
Env: newEnvironment,
VolumeMounts: []appv1.VolumeMount{
{
Name: "kubeconfig",
MountPath: "/etc/kubernetes/admin.conf",
},
},
},
},
Volumes: []appv1.Volume{
{
Name: "kubeconfig",
VolumeSource: appv1.VolumeSource{
HostPath: &appv1.HostPathVolumeSource{
Path: "/etc/kubernetes/admin.conf",
},
},
},
},
HostNetwork: true,
},
}
b, _ := yaml.Marshal(newManifest)
return string(b)
}

3
pkg/kubevip/config_types.go
View File

@@ -5,6 +5,9 @@ import "net/url"
// Config defines all of the settings for the Virtual IP / Load-balancer
type Config struct {
// EnableLeaderElection will use the Kubernetes leader election algorithim
EnableLeaderElection bool `yaml:"enableLeaderElection"`
// LocalPeer is the configuration of this host
LocalPeer RaftPeer `yaml:"localPeer"`