From 0c194d8179e6521b510d640c3618c0718e89a46e Mon Sep 17 00:00:00 2001 From: ston1th Date: Sat, 11 Dec 2021 18:58:58 +0100 Subject: [PATCH] added TODO --- TODO | 16 +--- pkg/etcd/etcd.go | 227 ----------------------------------------------- 2 files changed, 3 insertions(+), 240 deletions(-) delete mode 100644 pkg/etcd/etcd.go diff --git a/TODO b/TODO index 1198094..7e96b83 100644 --- a/TODO +++ b/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 diff --git a/pkg/etcd/etcd.go b/pkg/etcd/etcd.go deleted file mode 100644 index 3de7f08..0000000 --- a/pkg/etcd/etcd.go +++ /dev/null @@ -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) -}