diff --git a/cmd/vipman/main.go b/cmd/vipman/main.go index 1cb66e7..fb93d8a 100644 --- a/cmd/vipman/main.go +++ b/cmd/vipman/main.go @@ -66,47 +66,7 @@ func main() { sigs := make(chan os.Signal) signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) <-sigs - log.Info("stopping vipman") + log.Info("vipman shutdown") c.Stop() - //DeleteIPs(n, cfg.VirtualIPs) + log.Info("vipman stopped") } - -/* - c, err := raft.NewCluster(cluster.Callbacks{ - Leader: func(ctx context.Context, kv cluster.KV) { - ticker := time.NewTicker(time.Second * 10) - for { - s := kv.Get("state") - if s == nil { - s = []byte{0} - } else { - s[0] = s[0] + 1 - } - kv.Set("state", s) - s = kv.Get("state") - klog.Infof("[%s] l state: %d", cfg.LocalPeer.ID, s[0]) - select { - case <-ctx.Done(): - klog.Infof("[%s] leading canceled", cfg.LocalPeer.ID) - return - case <-ticker.C: - } - } - }, - Follower: func(ctx context.Context, kv cluster.KV) { - ticker := time.NewTicker(time.Second * 2) - for { - s := kv.Get("state") - if s != nil { - klog.Infof("[%s] f state: %d", cfg.LocalPeer.ID, s[0]) - } - select { - case <-ctx.Done(): - klog.Infof("[%s] following canceled", cfg.LocalPeer.ID) - return - case <-ticker.C: - } - } - }, - }) -*/ diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index 599da12..ca755df 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -15,7 +15,7 @@ type Cluster interface { } type KV interface { - Get(k string) []byte + Get(k string) ([]byte, error) Set(k string, v []byte) error Delete(k string) error } @@ -29,4 +29,5 @@ type CallbackContext interface { type Callbacks struct { Leader func(context.Context, CallbackContext) Follower func(context.Context, CallbackContext) + Cleanup func(context.Context, CallbackContext) } diff --git a/pkg/controller/vip.go b/pkg/controller/vip.go index 353882c..eb296d8 100644 --- a/pkg/controller/vip.go +++ b/pkg/controller/vip.go @@ -63,6 +63,9 @@ func NewVIPController(cfg *config.Config) (callbacks cluster.Callbacks, err erro } } }, + Cleanup: func(ctx context.Context, cc cluster.CallbackContext) { + deleteIPs(cc, n, cfg.VirtualIPs) + }, } return } diff --git a/pkg/etcd/cluster.go b/pkg/etcd/cluster.go index 10d8897..2aa3521 100644 --- a/pkg/etcd/cluster.go +++ b/pkg/etcd/cluster.go @@ -1,6 +1,7 @@ package etcd import ( + "context" "errors" "git.giftfish.de/ston1th/vipman/pkg/cluster" @@ -9,16 +10,19 @@ import ( "github.com/go-logr/logr" ) +var ErrKeyNotFound = errors.New("key not found") + type Cluster struct { logr.Logger - cluster.KV cli *clientv3.Client + ctx context.Context stepdown chan struct{} stop chan struct{} done chan struct{} callbacks cluster.Callbacks + kvPrefix string id string keyID string session bool @@ -28,13 +32,36 @@ type Cluster struct { func NewCluster(log logr.Logger) cluster.Cluster { return &Cluster{ Logger: log, - KV: &kv{}, + ctx: context.Background(), stepdown: make(chan struct{}, 1), stop: make(chan struct{}, 1), done: make(chan struct{}, 1), } } +func (c *Cluster) Get(k string) (v []byte, err error) { + r, err := c.cli.Get(c.ctx, c.kvPrefix+"/"+k) + if err != nil { + return + } + if len(r.Kvs) == 0 { + err = ErrKeyNotFound + return + } + v = r.Kvs[0].Value + return +} + +func (c *Cluster) Set(k string, v []byte) (err error) { + _, err = c.cli.Put(c.ctx, c.kvPrefix+"/"+k, string(v)) + return +} + +func (c *Cluster) Delete(k string) (err error) { + _, err = c.cli.Delete(c.ctx, c.kvPrefix+"/"+k) + return +} + func (c *Cluster) SetCallbacks(callbacks cluster.Callbacks) error { c.callbacks = callbacks return c.checkConfig() diff --git a/pkg/etcd/etcd.go b/pkg/etcd/etcd.go index 9c70479..f5df23a 100644 --- a/pkg/etcd/etcd.go +++ b/pkg/etcd/etcd.go @@ -73,16 +73,18 @@ func (c *Cluster) Start(etcdcfg *config.Config) error { c.session = true prefix := cfg.Prefix + "/" + cfg.ClusterName + c.kvPrefix = prefix + "/kv" + ePrefix := prefix + "/e" c.id = fmt.Sprintf("%x", s.Lease()) - c.keyID = fmt.Sprintf("%s/%s", prefix, c.id) + c.keyID = ePrefix + "/" + c.id - e := concurrency.NewElection(s, prefix) + e := concurrency.NewElection(s, ePrefix) ctxCampaign, cancelCampaign := context.WithCancel(context.Background()) go func() { errc := make(chan error, 1) for { go func() { - errc <- e.Campaign(ctxCampaign, "leader") + errc <- e.Campaign(ctxCampaign, "") }() select { case err := <-errc: @@ -141,12 +143,16 @@ func (c *Cluster) Start(etcdcfg *config.Config) error { e.Resign(ctxResign) case <-c.stop: t.Stop() + cancel() cancelObserve() cancelCampaign() - cancel() if leading { - ctxResign, _ := context.WithTimeout(context.Background(), time.Second*2) - e.Resign(ctxResign) + 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) } s.Close() close(c.done) diff --git a/pkg/etcd/kv.go b/pkg/etcd/kv.go deleted file mode 100644 index f229335..0000000 --- a/pkg/etcd/kv.go +++ /dev/null @@ -1,18 +0,0 @@ -package etcd - -//TODO - -type kv struct { -} - -func (kv *kv) Get(k string) []byte { - return nil -} - -func (kv *kv) Set(k string, v []byte) error { - return nil -} - -func (kv *kv) Delete(k string) error { - return nil -} diff --git a/pkg/raft/kv.go b/pkg/raft/kv.go index baaed80..8c14923 100644 --- a/pkg/raft/kv.go +++ b/pkg/raft/kv.go @@ -18,10 +18,10 @@ type kv struct { r *raft.Raft } -func (kv *kv) Get(k string) []byte { +func (kv *kv) Get(k string) ([]byte, error) { kv.mu.Lock() defer kv.mu.Unlock() - return kv.m[k] + return kv.m[k], nil } func (kv *kv) Set(k string, v []byte) error { diff --git a/pkg/raft/raft.go b/pkg/raft/raft.go index 59fa2b7..afbdad9 100644 --- a/pkg/raft/raft.go +++ b/pkg/raft/raft.go @@ -155,6 +155,10 @@ func (c *Cluster) Start(raftcfg *config.Config) error { if leading { c.r.LeadershipTransfer().Error() } + if c.callbacks.Cleanup != nil { + ctx, _ := context.WithTimeout(context.Background(), time.Second*2) + c.callbacks.Cleanup(ctx, c) + } close(c.done) return nil }