added TODO
This commit is contained in:
parent
eb2ed92605
commit
0c194d8179
2 changed files with 3 additions and 240 deletions
16
TODO
16
TODO
|
|
@ -1,15 +1,5 @@
|
|||
errors:
|
||||
use new etcd impl
|
||||
|
||||
1. etcd stopped
|
||||
2. crtl-C
|
||||
bla etcd "ts"="2021-11-07 23:45:03.517326" "caller"={"file":"etcd.go","line":94} "msg"="campaign error" "error"="rpc error: code = Unavailable desc = error reading from server: EOF"
|
||||
bla etcd "ts"="2021-11-07 23:45:03.518082" "caller"={"file":"loadbalancer.go","line":118} "level"=0 "msg"="stopping haproxy" "service"="haproxy"
|
||||
bla etcd "ts"="2021-11-07 23:45:03.529380" "caller"={"file":"loadbalancer.go","line":123} "level"=0 "msg"="leading canceled" "id"="6a6c7cfc8dd69b62"
|
||||
^Cbla main "ts"="2021-11-07 23:45:08.926006" "caller"={"file":"main.go","line":118} "level"=0 "msg"="haproxy-lb shutdown" "signal"="interrupt"
|
||||
panic: send on closed channel
|
||||
remove raft
|
||||
|
||||
goroutine 1 [running]:
|
||||
git.giftfish.de/ston1th/haproxy-lb/pkg/etcd.(*Cluster).Stop(0xc00017d340)
|
||||
git.giftfish.de/ston1th/haproxy-lb/pkg/etcd/etcd.go:213 +0x33
|
||||
main.main()
|
||||
git.giftfish.de/ston1th/haproxy-lb/cmd/haproxy-lb/main.go:128 +0xebb
|
||||
add new etcd config
|
||||
|
|
|
|||
227
pkg/etcd/etcd.go
227
pkg/etcd/etcd.go
|
|
@ -1,227 +0,0 @@
|
|||
package etcd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"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"
|
||||
"go.etcd.io/etcd/client/v3/concurrency"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
)
|
||||
|
||||
func (c *Cluster) Start(etcdcfg *config.Config) error {
|
||||
err := c.checkConfig()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cfg := etcdcfg.Cluster.Etcd
|
||||
var level zapcore.Level
|
||||
err = level.Set(cfg.LogLevel)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
al := zap.NewAtomicLevel()
|
||||
al.SetLevel(level)
|
||||
var tlsConfig *tls.Config
|
||||
if cfg.TLS != nil {
|
||||
ca, err := config.LoadCertPool(cfg.TLS.CA)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tlsConfig = &tls.Config{
|
||||
MinVersion: tls.VersionTLS12,
|
||||
InsecureSkipVerify: cfg.TLS.Insecure,
|
||||
RootCAs: ca,
|
||||
}
|
||||
}
|
||||
|
||||
c.cli, err = clientv3.New(clientv3.Config{
|
||||
Endpoints: cfg.Endpoints,
|
||||
TLS: tlsConfig,
|
||||
DialTimeout: time.Second * 5,
|
||||
Username: cfg.Username,
|
||||
Password: cfg.Password,
|
||||
LogConfig: &zap.Config{
|
||||
Level: al,
|
||||
Encoding: "console",
|
||||
DisableCaller: true,
|
||||
DisableStacktrace: true,
|
||||
EncoderConfig: zap.NewDevelopmentEncoderConfig(),
|
||||
OutputPaths: []string{"stderr"},
|
||||
ErrorOutputPaths: []string{"stderr"},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
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(sessionCtx),
|
||||
concurrency.WithTTL(10),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.session = true
|
||||
|
||||
prefix := cfg.Prefix + "/" + cfg.ClusterName
|
||||
c.kvPrefix = prefix + "/kv"
|
||||
ePrefix := prefix + "/e"
|
||||
c.id = fmt.Sprintf("%x", s.Lease())
|
||||
c.keyID = ePrefix + "/" + c.id
|
||||
|
||||
e := concurrency.NewElection(s, ePrefix)
|
||||
campaignCtx, campaignCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
errc := make(chan error, 1)
|
||||
for {
|
||||
go func() {
|
||||
errc <- e.Campaign(campaignCtx, "")
|
||||
}()
|
||||
select {
|
||||
case err := <-errc:
|
||||
if err != nil {
|
||||
log.Error(err, "campaign error")
|
||||
c.Fatal(cluster.ErrRestart)
|
||||
}
|
||||
case <-campaignCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
t := time.NewTicker(time.Second * 2)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
observeCtx, observeCancel := context.WithCancel(context.Background())
|
||||
leading := false
|
||||
following := false
|
||||
leaderChan := c.observe(observeCtx, e.Observe(observeCtx))
|
||||
for {
|
||||
if !leading && c.leaderID(e.Leader(observeCtx)) == c.keyID {
|
||||
leading = true
|
||||
go c.leader(ctx)
|
||||
} else if !following && !leading {
|
||||
following = true
|
||||
go c.follower(ctx)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-t.C:
|
||||
case leader := <-leaderChan:
|
||||
if leader {
|
||||
if following {
|
||||
cancel()
|
||||
following = false
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
}
|
||||
if !leading {
|
||||
leading = true
|
||||
go c.leader(ctx)
|
||||
}
|
||||
} else {
|
||||
if leading {
|
||||
cancel()
|
||||
leading = false
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
}
|
||||
if !following {
|
||||
following = true
|
||||
go c.follower(ctx)
|
||||
}
|
||||
}
|
||||
case <-c.stepdown:
|
||||
cancel()
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
resignCtx, _ := context.WithTimeout(context.Background(), time.Second*2)
|
||||
e.Resign(resignCtx)
|
||||
case <-c.stop:
|
||||
t.Stop()
|
||||
cancel()
|
||||
observeCancel()
|
||||
campaignCancel()
|
||||
if leading {
|
||||
ctx, _ := context.WithTimeout(context.Background(), time.Second*2)
|
||||
e.Resign(ctx)
|
||||
}
|
||||
if c.callbacks.Cleanup != nil {
|
||||
c.callbacks.Cleanup(context.Background(), c)
|
||||
}
|
||||
s.Close()
|
||||
close(c.done)
|
||||
return c.fatal
|
||||
}
|
||||
}
|
||||
return c.fatal
|
||||
}
|
||||
|
||||
func (c *Cluster) leaderID(r *clientv3.GetResponse, err error) string {
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
if len(r.Kvs) == 0 {
|
||||
return ""
|
||||
}
|
||||
return string(r.Kvs[0].Key)
|
||||
}
|
||||
|
||||
func (c *Cluster) observe(ctx context.Context, resp <-chan clientv3.GetResponse) <-chan bool {
|
||||
leader := make(chan bool)
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case r := <-resp:
|
||||
if c.leaderID(&r, nil) == c.keyID {
|
||||
leader <- true
|
||||
} else {
|
||||
leader <- false
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
return leader
|
||||
}
|
||||
|
||||
func (c *Cluster) Stepdown() {
|
||||
select {
|
||||
case c.stepdown <- struct{}{}:
|
||||
case <-time.After(time.Second * 2):
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cluster) Fatal(err error) {
|
||||
c.fatal = err
|
||||
c.Stop()
|
||||
}
|
||||
|
||||
func (c *Cluster) Stop() {
|
||||
if !c.session {
|
||||
c.sessionCancel()
|
||||
return
|
||||
}
|
||||
c.stop <- struct{}{}
|
||||
close(c.stop)
|
||||
<-c.done
|
||||
c.cli.Close()
|
||||
}
|
||||
|
||||
func (c *Cluster) leader(ctx context.Context) {
|
||||
defer util.HandleCrash()
|
||||
c.callbacks.Leader(ctx, c)
|
||||
}
|
||||
|
||||
func (c *Cluster) follower(ctx context.Context) {
|
||||
defer util.HandleCrash()
|
||||
c.callbacks.Follower(ctx, c)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue