ston1th/haproxy-lb
0
0
Fork 0
Code Issues Releases Activity

new logger and etcd fixes

Browse source
This commit is contained in:
ston1th 2021-10-09 16:52:06 +02:00
commit cbd4f38cca
11 changed files with 248 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

4
pkg/cluster/cluster.go
View file

@ -12,6 +12,7 @@ var (
ErrNotLeader = errors.New("not leader")
ErrKeyNotFound = errors.New("key not found")
ErrPrefixNotFound = errors.New("prefix not found")
ErrRestart = errors.New("restart")
)
type Cluster interface {
@ -19,6 +20,7 @@ type Cluster interface {
SetCallbacks(Callbacks) error
Stepdown()
Stop()
Reset()
}
type KV interface {
@ -29,8 +31,8 @@ type KV interface {
}
type CallbackContext interface {
logr.Logger
KV
Logger() logr.Logger
ID() string
Fatal(error)
}

54
pkg/controller/loadbalancer.go
View file

@ -27,16 +27,17 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call
callbacks = cluster.Callbacks{
Leader: func(ctx context.Context, cc cluster.CallbackContext) {
db := db.New(cc)
cc.V(2).Info("starting haproxy", "service", cfg.HAProxyService)
log := cc.Logger()
log.Info("starting haproxy", "service", cfg.HAProxyService)
err := ha.Start(ctx)
if err != nil {
cc.Error(err, "error starting haproxy")
log.Error(err, "error starting haproxy")
cc.Fatal(err)
return
}
err = srv.UpdateDB(db)
if err != nil {
cc.Error(err, "error initialising api server as leader")
log.Error(err, "error initialising api server as leader")
cc.Fatal(err)
return
}
@ -45,56 +46,61 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call
go func() {
err := exec.CommandContext(ctx, cfg.LeaderHook).Run()
if err != nil {
cc.Error(err, "error running hook", "leaderHook", cfg.LeaderHook)
log.Error(err, "error running hook", "leaderHook", cfg.LeaderHook)
}
}()
}
for {
cc.Info("leading", "id", cc.ID())
if ctx.Err() == context.Canceled {
t.Stop()
log.Info("leading canceled", "id", cc.ID())
return
}
log.Info("leading", "id", cc.ID())
ips, err := db.GetDeletedIPs()
if err != nil && err != cluster.ErrPrefixNotFound {
cc.Error(err, "error reading deleted ips list")
log.Error(err, "error reading deleted ips list")
<-t.C
continue
}
deleteIPs(cc, n, ips)
deleteIPs(log, n, ips)
ips, err = db.GetIPs()
if err != nil && err != cluster.ErrPrefixNotFound {
cc.Error(err, "error reading ips list")
log.Error(err, "error reading ips list")
<-t.C
continue
}
addIPs(cc, n, ips)
addIPs(log, n, ips)
lbs, err := db.GetLBs("")
if err != nil && err != cluster.ErrPrefixNotFound {
cc.Error(err, "error reading lb list")
log.Error(err, "error reading lb list")
<-t.C
continue
}
lbcfg, err := haproxy.NewConfig(lbs)
if err != nil {
cc.Error(err, "error reading haproxy config")
log.Error(err, "error reading haproxy config")
<-t.C
continue
}
err = ha.UpdateConfig(ctx, lbcfg)
if err != nil {
cc.Error(err, "error updating haproxy config")
log.Error(err, "error updating haproxy config")
<-t.C
continue
}
select {
case <-ctx.Done():
t.Stop()
deleteIPs(cc, n, ips)
cc.V(2).Info("stopping haproxy", "service", cfg.HAProxyService)
deleteIPs(log, n, ips)
log.Info("stopping haproxy", "service", cfg.HAProxyService)
err = ha.Stop(context.Background())
if err != nil {
cc.Error(err, "error stopping haproxy")
log.Error(err, "error stopping haproxy")
}
cc.Info("leading canceled", "id", cc.ID())
log.Info("leading canceled", "id", cc.ID())
return
case <-t.C:
}
@ -102,9 +108,10 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call
},
Follower: func(ctx context.Context, cc cluster.CallbackContext) {
db := db.New(cc)
log := cc.Logger()
err := srv.UpdateDB(db)
if err != nil {
cc.Error(err, "error initialising api server as follower")
log.Error(err, "error initialising api server as follower")
cc.Fatal(err)
return
}
@ -113,23 +120,28 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call
go func() {
err := exec.CommandContext(ctx, cfg.FollowerHook).Run()
if err != nil {
cc.Error(err, "error running hook", "followerHook", cfg.FollowerHook)
log.Error(err, "error running hook", "followerHook", cfg.FollowerHook)
}
}()
}
for {
cc.Info("following", "id", cc.ID())
if ctx.Err() == context.Canceled {
t.Stop()
log.Info("following canceled", "id", cc.ID())
return
}
log.Info("following", "id", cc.ID())
select {
case <-ctx.Done():
t.Stop()
cc.Info("following canceled", "id", cc.ID())
log.Info("following canceled", "id", cc.ID())
return
case <-t.C:
}
}
},
Cleanup: func(ctx context.Context, cc cluster.CallbackContext) {
deleteIPs(cc, n, n.GetIPs())
deleteIPs(log, n, n.GetIPs())
},
}
return

38
pkg/db/db.go
View file

@ -1,7 +1,9 @@
package db
import (
"errors"
"sync"
"time"
schemav1 "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema"
"git.giftfish.de/ston1th/haproxy-lb/pkg/cluster"
@ -34,17 +36,37 @@ func (db *DB) GetIPs() (ips []string, err error) {
return
}
func (db *DB) GetDeletedIPs() (ips []string, err error) {
m, err := db.kv.GetPrefix(delPrefix)
if err != nil {
return
}
for k := range m {
ips = append(ips, k)
db.kv.Delete(delPrefix + k)
}
err = timeout(func() error {
m, err := db.kv.GetPrefix(delPrefix)
if err != nil {
return err
}
for k := range m {
ips = append(ips, k)
db.kv.Delete(delPrefix + k)
}
return nil
})
return
}
func timeout(f func() error) error {
return timeoutWithDuration(f, time.Second*3)
}
func timeoutWithDuration(f func() error, d time.Duration) error {
c := make(chan error, 1)
go func() {
e := f()
c <- e
}()
select {
case err := <-c:
return err
case <-time.After(d):
return errors.New("timeout")
}
}
func (db *DB) GetCIDR(name string) (cidr string, err error) {
b, err := db.GetLB(name)
if err != nil {

27
pkg/etcd/cluster.go
View file

@ -11,7 +11,7 @@ import (
)
type Cluster struct {
logr.Logger
log logr.Logger
cli *clientv3.Client
ctx context.Context
@ -24,18 +24,23 @@ type Cluster struct {
id string
keyID string
session bool
cancelSession func()
sessionCancel func()
fatal error
}
func NewCluster(log logr.Logger) cluster.Cluster {
return &Cluster{
Logger: log,
ctx: context.Background(),
stepdown: make(chan struct{}, 1),
stop: make(chan struct{}, 1),
done: make(chan struct{}, 1),
}
c := &Cluster{log: log}
c.Reset()
return c
}
func (c *Cluster) Reset() {
c.ctx = context.Background()
c.stepdown = make(chan struct{}, 1)
c.stop = make(chan struct{}, 1)
c.done = make(chan struct{}, 1)
c.session = false
c.fatal = nil
}
func (c *Cluster) Get(k string) (v []byte, err error) {
@ -83,6 +88,10 @@ func (c *Cluster) SetCallbacks(callbacks cluster.Callbacks) error {
return c.checkConfig()
}
func (c *Cluster) Logger() logr.Logger {
return c.log
}
func (c *Cluster) ID() string {
return c.id
}

40
pkg/etcd/etcd.go
View file

@ -6,6 +6,7 @@ import (
"fmt"
"time"
"git.giftfish.de/ston1th/haproxy-lb/pkg/cluster"
"git.giftfish.de/ston1th/haproxy-lb/pkg/config"
"git.giftfish.de/ston1th/haproxy-lb/pkg/util"
clientv3 "go.etcd.io/etcd/client/v3"
@ -60,10 +61,12 @@ func (c *Cluster) Start(etcdcfg *config.Config) error {
return err
}
ctxSsession, cancelSession := context.WithCancel(context.Background())
c.cancelSession = cancelSession
log := c.Logger()
log.Info("waiting for etcd")
sessionCtx, sessionCancel := context.WithCancel(context.Background())
c.sessionCancel = sessionCancel
s, err := concurrency.NewSession(c.cli,
concurrency.WithContext(ctxSsession),
concurrency.WithContext(sessionCtx),
concurrency.WithTTL(10),
)
if err != nil {
@ -78,32 +81,33 @@ func (c *Cluster) Start(etcdcfg *config.Config) error {
c.keyID = ePrefix + "/" + c.id
e := concurrency.NewElection(s, ePrefix)
ctxCampaign, cancelCampaign := context.WithCancel(context.Background())
campaignCtx, campaignCancel := context.WithCancel(context.Background())
go func() {
errc := make(chan error, 1)
for {
go func() {
errc <- e.Campaign(ctxCampaign, "")
errc <- e.Campaign(campaignCtx, "")
}()
select {
case err := <-errc:
if err != nil {
time.Sleep(time.Second)
log.Error(err, "campaign error")
c.Fatal(cluster.ErrRestart)
}
case <-ctxCampaign.Done():
case <-campaignCtx.Done():
return
}
}
}()
t := time.NewTicker(time.Second)
t := time.NewTicker(time.Second * 2)
ctx, cancel := context.WithCancel(context.Background())
ctxObserve, cancelObserve := context.WithCancel(context.Background())
observeCtx, observeCancel := context.WithCancel(context.Background())
leading := false
following := false
leaderChan := c.observe(ctxObserve, e.Observe(ctxObserve))
leaderChan := c.observe(observeCtx, e.Observe(observeCtx))
for {
if !leading && c.leaderID(e.Leader(ctxObserve)) == c.keyID {
if !leading && c.leaderID(e.Leader(observeCtx)) == c.keyID {
leading = true
go c.leader(ctx)
} else if !following && !leading {
@ -138,20 +142,19 @@ func (c *Cluster) Start(etcdcfg *config.Config) error {
case <-c.stepdown:
cancel()
ctx, cancel = context.WithCancel(context.Background())
ctxResign, _ := context.WithTimeout(context.Background(), time.Second*2)
e.Resign(ctxResign)
resignCtx, _ := context.WithTimeout(context.Background(), time.Second*2)
e.Resign(resignCtx)
case <-c.stop:
t.Stop()
cancel()
cancelObserve()
cancelCampaign()
observeCancel()
campaignCancel()
if leading {
ctx, _ := context.WithTimeout(context.Background(), time.Second*2)
e.Resign(ctx)
}
if c.callbacks.Cleanup != nil {
ctx, _ := context.WithTimeout(context.Background(), time.Second*2)
c.callbacks.Cleanup(ctx, c)
c.callbacks.Cleanup(context.Background(), c)
}
s.Close()
close(c.done)
@ -204,9 +207,10 @@ func (c *Cluster) Fatal(err error) {
func (c *Cluster) Stop() {
if !c.session {
c.cancelSession()
c.sessionCancel()
return
}
c.stop <- struct{}{}
close(c.stop)
<-c.done
c.cli.Close()

28
pkg/raft/cluster.go
View file

@ -9,9 +9,9 @@ import (
)
type Cluster struct {
logr.Logger
cluster.KV
r *raft.Raft
log logr.Logger
r *raft.Raft
kvm *kv
stepdown chan struct{}
@ -24,17 +24,25 @@ type Cluster struct {
}
func NewCluster(log logr.Logger) cluster.Cluster {
c := &Cluster{log: log}
c.Reset()
return c
}
func (c *Cluster) Reset() {
kvm := &kv{
m: make(map[string][]byte),
}
return &Cluster{
Logger: log,
KV: kvm,
kvm: kvm,
stepdown: make(chan struct{}, 1),
stop: make(chan struct{}, 1),
done: make(chan struct{}, 1),
}
c.KV = kvm
c.kvm = kvm
c.stepdown = make(chan struct{}, 1)
c.stop = make(chan struct{}, 1)
c.done = make(chan struct{}, 1)
c.fatal = nil
}
func (c *Cluster) Logger() logr.Logger {
return c.log
}
func (c *Cluster) ID() string {

5
pkg/raft/raft.go
View file

@ -21,6 +21,7 @@ func (c *Cluster) Start(raftcfg *config.Config) error {
if err != nil {
return err
}
log := c.Logger()
cfg := raftcfg.Cluster.Raft
hcl := hclog.New(&hclog.LoggerOptions{
Name: "raft",
@ -39,14 +40,14 @@ func (c *Cluster) Start(raftcfg *config.Config) error {
timeout := 10 * time.Second
var transport *raft.NetworkTransport
if cfg.TLS != nil {
c.V(1).Info("setup", "transport", "tls")
log.V(1).Info("setup", "transport", "tls")
s, err := tls.NewStream(cfg)
if err != nil {
return err
}
transport = raft.NewNetworkTransportWithLogger(s, pool, timeout, hcl)
} else {
c.V(1).Info("setup", "transport", "tcp")
log.V(1).Info("setup", "transport", "tcp")
address, err := net.ResolveTCPAddr("tcp", cfg.Address)
if err != nil {
return err

14
pkg/util/crash.go
View file

@ -1,13 +1,19 @@
package util
import (
"errors"
"fmt"
"net/http"
"runtime"
"k8s.io/klog/v2"
"github.com/iand/logfmtr"
)
var reallyCrash = true
var (
reallyCrash = true
log = logfmtr.New()
panicErr = errors.New("observed a panic")
)
func HandleCrash() {
if r := recover(); r != nil {
@ -34,8 +40,8 @@ func logPanic(r interface{}) {
stacktrace := make([]byte, size)
stacktrace = stacktrace[:runtime.Stack(stacktrace, false)]
if _, ok := r.(string); ok {
klog.Errorf("Observed a panic: %s\n%s", r, stacktrace)
log.Error(panicErr, fmt.Sprintf("%s\n%s", r, stacktrace))
} else {
klog.Errorf("Observed a panic: %#v (%v)\n%s", r, r, stacktrace)
log.Error(panicErr, fmt.Sprintf("%#v (%v)\n%s", r, r, stacktrace))
}
}