From 585baea183e3ea7ba4e83783e112afb98f2ad595 Mon Sep 17 00:00:00 2001 From: ston1th Date: Sun, 15 Nov 2020 00:30:35 +0100 Subject: [PATCH] added raft over tls --- VERSION | 1 + cmd/vipman/main.go | 134 +++++++++++++++++++---- go.mod | 7 +- go.sum | 53 ++++++++- pkg/cluster/cluster.go | 44 +++----- pkg/cluster/marshaler.go | 21 ---- pkg/config/config.go | 111 +++++++++++++++---- pkg/raft/cluster.go | 36 +++++++ pkg/{cluster => raft}/fsm.go | 10 +- pkg/{cluster => raft}/kv.go | 7 +- pkg/{cluster => raft}/raft.go | 127 ++++++++++------------ pkg/raft/tls/stream.go | 89 ++++++++++++++++ pkg/vip/arp.go | 2 +- pkg/vip/arp_linux.go | 23 ++-- pkg/vip/vip.go | 195 +++++++++++++--------------------- vipman.yaml | 41 ++++--- 16 files changed, 582 insertions(+), 319 deletions(-) create mode 100644 VERSION delete mode 100644 pkg/cluster/marshaler.go create mode 100644 pkg/raft/cluster.go rename pkg/{cluster => raft}/fsm.go (85%) rename pkg/{cluster => raft}/kv.go (86%) rename pkg/{cluster => raft}/raft.go (57%) create mode 100644 pkg/raft/tls/stream.go diff --git a/VERSION b/VERSION new file mode 100644 index 0000000..49d5957 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +0.1 diff --git a/cmd/vipman/main.go b/cmd/vipman/main.go index 678d150..b15991d 100644 --- a/cmd/vipman/main.go +++ b/cmd/vipman/main.go @@ -4,20 +4,45 @@ import ( "context" "flag" "os" + "os/exec" "os/signal" + "path/filepath" "syscall" "time" "git.giftfish.de/ston1th/vipman/pkg/cluster" "git.giftfish.de/ston1th/vipman/pkg/config" + "git.giftfish.de/ston1th/vipman/pkg/raft" + "git.giftfish.de/ston1th/vipman/pkg/vip" "k8s.io/klog" "k8s.io/klog/klogr" ) var ( configFile string + version string ) +func AddIPs(n *vip.Network, cidrs []string) error { + for _, cidr := range cidrs { + err := n.AddIP(cidr) + if err != nil { + klog.Infof("error adding ip %s to interface %s: %s", cidr, n.Interface(), err) + } + } + return nil +} + +func DeleteIPs(n *vip.Network, cidrs []string) error { + for _, cidr := range cidrs { + err := n.DeleteIP(cidr) + if err != nil { + klog.Infof("error deleting ip %s from interface %s: %s", cidr, n.Interface(), err) + } + } + return nil +} + func main() { klog.InitFlags(nil) flag.StringVar(&configFile, "config", "vipman.yaml", "vipman config file") @@ -26,8 +51,95 @@ func main() { if err != nil { klog.Fatalf("Initialization failed: %s", err) } - c, err := cluster.InitCluster(cluster.Callbacks{ - Leader: func(ctx context.Context, kv *cluster.KV) { + n, err := vip.NewNetworkWithLabel(cfg.Interface, cfg.IPLabel) + if err != nil { + klog.Fatalf("Initialization failed: %s", err) + } + cc := cfg.Cluster.Raft + + c, err := raft.NewCluster(cluster.Callbacks{ + Leader: func(ctx context.Context, _ cluster.KV) { + t := time.NewTicker(time.Second * 10) + if cfg.LeaderHook != "" { + if !filepath.IsAbs(cfg.LeaderHook) { + klog.Infof("error running leaderHook %s: path is not absolute", cfg.LeaderHook) + } else { + go func() { + err := exec.CommandContext(ctx, cfg.LeaderHook).Run() + if err != nil { + klog.Infof("error running leaderHook %s: %s", cfg.LeaderHook, err) + } + }() + } + } + for { + klog.Infof("[%s] leading", cc.ID) + AddIPs(n, cfg.VirtualIPs) + select { + case <-ctx.Done(): + t.Stop() + DeleteIPs(n, cfg.VirtualIPs) + klog.Infof("[%s] leading canceled", cc.ID) + return + case <-t.C: + } + } + }, + Follower: func(ctx context.Context, _ cluster.KV) { + t := time.NewTicker(time.Second * 10) + if cfg.FollowerHook != "" { + if !filepath.IsAbs(cfg.FollowerHook) { + klog.Infof("error running followerHook %s: path is not absolute", cfg.FollowerHook) + } else { + go func() { + err := exec.CommandContext(ctx, cfg.FollowerHook).Run() + if err != nil { + klog.Infof("error running FollowerHook %s: %s", cfg.FollowerHook, err) + } + }() + } + } + for { + klog.Infof("[%s] following", cc.ID) + //DeleteIPs(n, cfg.VirtualIPs) + select { + case <-ctx.Done(): + t.Stop() + klog.Infof("[%s] following canceled", cc.ID) + return + case <-t.C: + } + } + }, + }) + if err != nil { + klog.Fatalf("Initialization failed: %s", err) + } + log := klogr.New() + go func() { + err = c.Start(log, cfg) + if err != nil { + klog.Fatalf("Initialization failed: %s", err) + } + }() + go func() { + sig := make(chan os.Signal) + signal.Notify(sig, syscall.SIGHUP) + for { + <-sig + c.Stepdown() + } + }() + sigs := make(chan os.Signal) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + <-sigs + c.Stop() + DeleteIPs(n, cfg.VirtualIPs) +} + +/* + 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") @@ -47,7 +159,7 @@ func main() { } } }, - Follower: func(ctx context.Context, kv *cluster.KV) { + Follower: func(ctx context.Context, kv cluster.KV) { ticker := time.NewTicker(time.Second * 2) for { s := kv.Get("state") @@ -63,18 +175,4 @@ func main() { } }, }) - if err != nil { - klog.Fatalf("Initialization failed: %s", err) - } - log := klogr.New() - go func() { - err = c.StartRaftCluster(log, cfg) - if err != nil { - klog.Fatalf("Initialization failed: %s", err) - } - }() - sigs := make(chan os.Signal) - signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) - <-sigs - c.Stop() -} +*/ diff --git a/go.mod b/go.mod index cebb419..18e301f 100644 --- a/go.mod +++ b/go.mod @@ -3,9 +3,14 @@ module git.giftfish.de/ston1th/vipman go 1.15 require ( + git.giftfish.de/ston1th/raftbbolt v0.0.0-20201114165203-b85c94bfc9b0 + github.com/fatih/color v1.10.0 // indirect github.com/go-logr/logr v0.3.0 + github.com/hashicorp/go-hclog v0.15.0 github.com/hashicorp/raft v1.2.0 - github.com/hashicorp/raft-boltdb v0.0.0-20191021154308-4207f1bf0617 + github.com/vishvananda/netlink v1.1.0 + golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba // indirect gopkg.in/yaml.v2 v2.3.0 k8s.io/klog v1.0.0 + k8s.io/klog/v2 v2.4.0 ) diff --git a/go.sum b/go.sum index 7234155..f788287 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +git.giftfish.de/ston1th/raftbbolt v0.0.0-20201114165203-b85c94bfc9b0 h1:DU0PVdA2EHz2LFGEdkwgJopL9HQh5nFC9GJ7oqzZ300= +git.giftfish.de/ston1th/raftbbolt v0.0.0-20201114165203-b85c94bfc9b0/go.mod h1:9dF/aHYkXts6/6oVyJ+vzXCn7NURDaoREaanX0aspx0= github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878 h1:EFSB7Zo9Eg91v7MJPVsifUysc/wPdN+NOnVe6bWbdBM= github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg= @@ -7,31 +9,50 @@ github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx2 github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag= github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp5jckzBHf4XRpQvBOLI+I= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys= +github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= +github.com/fatih/color v1.10.0 h1:s36xzo75JdqLaaWoiEHk767eHiwo0598uUxyfiPkDsg= +github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM= github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas= +github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU= github.com/go-logr/logr v0.3.0 h1:q4c+kbcR0d5rSurhBR8dIgieOaYpXtsdTYfx22Cu6rs= github.com/go-logr/logr v0.3.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= github.com/hashicorp/go-hclog v0.9.1 h1:9PZfAcVEvez4yhLH2TBU64/h/z4xlFI80cWXRrxuKuM= github.com/hashicorp/go-hclog v0.9.1/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= +github.com/hashicorp/go-hclog v0.15.0 h1:qMuK0wxsoW4D0ddCCYwPSTm4KQv1X1ke3WmPWZ0Mvsk= +github.com/hashicorp/go-hclog v0.15.0/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ= github.com/hashicorp/go-immutable-radix v1.0.0 h1:AKDB1HM5PWEA7i4nhcpwOrO2byshxBjXVn/J/3+z5/0= github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= github.com/hashicorp/go-msgpack v0.5.5 h1:i9R9JSrqIz0QVLz3sz+i3YJdT7TTSLcfLLzJi9aZTuI= github.com/hashicorp/go-msgpack v0.5.5/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM= +github.com/hashicorp/go-msgpack v1.1.5 h1:9byZdVjKTe5mce63pRVNP1L7UAmdHOTEMGehn6KvJWs= +github.com/hashicorp/go-msgpack v1.1.5/go.mod h1:gWVc3sv/wbDmR3rQsj1CAktEZzoz1YNK9NfGLXJ69/4= github.com/hashicorp/go-retryablehttp v0.5.3/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs= +github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCSnM= github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= -github.com/hashicorp/raft v1.1.0/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM= github.com/hashicorp/raft v1.2.0 h1:mHzHIrF0S91d3A7RPBvuqkgB4d/7oFJZyvf1Q4m7GA0= github.com/hashicorp/raft v1.2.0/go.mod h1:vPAJM8Asw6u8LxC3eJCUZmRP/E4QmUGE1R7g7k8sG/8= github.com/hashicorp/raft-boltdb v0.0.0-20171010151810-6e5ba93211ea/go.mod h1:pNv7Wc3ycL6F5oOWn+tPGo2gWD4a5X+yp/ntwdKLjRk= -github.com/hashicorp/raft-boltdb v0.0.0-20191021154308-4207f1bf0617 h1:CJDRE/2tBNFOrcoexD2nvTRbQEox3FDxl4NxIezp1b8= -github.com/hashicorp/raft-boltdb v0.0.0-20191021154308-4207f1bf0617/go.mod h1:aUF6HQr8+t3FC/ZHAC+pZreUBhTaxumuu3L+d37uRxk= +github.com/mattn/go-colorable v0.1.4 h1:snbPLB8fVfU9iwbbo30TPtbLRzwWu6aJS6Xh4eaaviA= +github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE= +github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8= +github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= +github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s= +github.com/mattn/go-isatty v0.0.10 h1:qxFzApOv4WsAL965uUPIsXzAKCZxN2p9UqdhFS4ZW10= +github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84= +github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY= +github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY= github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM= github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= @@ -39,14 +60,38 @@ github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7q github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM= +github.com/vishvananda/netlink v1.1.0 h1:1iyaYNBLmP6L0220aDnYQpo1QEV4t4hJ+xEEhhJH8j0= +github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE= +github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df h1:OviZH7qLw/7ZovXvuNyL3XQl8UFofeikI1NW1Gypu7k= +github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU= +go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0= +go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190523142557-0e01d883c5c5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5 h1:LfCXLvNmTYH9kEmVgqbnsWfruoXZIrh4YBgqVHtDvw0= +golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba h1:xmhUJGQGbxlod18iJGqVEp9cHIPLl7QiX2aA3to708s= +golang.org/x/sys v0.0.0-20201113233024-12cec1faf1ba/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/tools v0.0.0-20190424220101-1e8e1cfdf96b/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU= gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8= k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I= +k8s.io/klog/v2 v2.4.0 h1:7+X0fUguPyrKEC4WjH8iGDg3laWgMo5tMnRTIGTTxGQ= +k8s.io/klog/v2 v2.4.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y= diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index 9e58335..b4eaef1 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -2,42 +2,24 @@ package cluster import ( "context" - "errors" - "github.com/hashicorp/raft" + "git.giftfish.de/ston1th/vipman/pkg/config" + "github.com/go-logr/logr" ) -// Cluster - The Cluster object manages the state of the cluster for a particular node -type Cluster struct { - kv *KV - r *raft.Raft +type Cluster interface { + Start(logr.Logger, *config.Config) error + Stepdown() + Stop() +} - stop chan struct{} - done chan struct{} - Callbacks Callbacks +type KV interface { + Get(k string) []byte + Set(k string, v []byte) error + Delete(k string) error } type Callbacks struct { - Leader func(context.Context, *KV) - Follower func(context.Context, *KV) -} - -// InitCluster - Will attempt to initialise all of the required settings for the cluster -func InitCluster(callbacks Callbacks) (*Cluster, error) { - if callbacks.Leader == nil { - return nil, errors.New("Leader is nil") - } - if callbacks.Follower == nil { - return nil, errors.New("Follower is nil") - } - c := &Cluster{ - kv: &KV{ - m: make(map[string][]byte), - }, - stop: make(chan struct{}, 1), - done: make(chan struct{}, 1), - Callbacks: callbacks, - } - - return c, nil + Leader func(context.Context, KV) + Follower func(context.Context, KV) } diff --git a/pkg/cluster/marshaler.go b/pkg/cluster/marshaler.go deleted file mode 100644 index ce625bf..0000000 --- a/pkg/cluster/marshaler.go +++ /dev/null @@ -1,21 +0,0 @@ -package cluster - -import ( - "bytes" - gogob "encoding/gob" -) - -type gobMarshaler struct{} - -func (gobMarshaler) Marshal(v interface{}) (b []byte, err error) { - buf := new(bytes.Buffer) - err = gogob.NewEncoder(buf).Encode(v) - b = buf.Bytes() - return -} - -func (gobMarshaler) Unmarshal(data []byte, v interface{}) error { - return gogob.NewDecoder(bytes.NewBuffer(data)).Decode(v) -} - -var gob = gobMarshaler{} diff --git a/pkg/config/config.go b/pkg/config/config.go index 791ef7f..9ed8022 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -2,8 +2,15 @@ package config import ( "errors" - "gopkg.in/yaml.v2" + "fmt" "os" + + "gopkg.in/yaml.v2" +) + +const ( + RaftDir = "raft" + RaftPort = 7001 ) func ParseFile(file string) (cfg *Config, err error) { @@ -16,33 +23,99 @@ func ParseFile(file string) (cfg *Config, err error) { } cfg = new(Config) err = yaml.NewDecoder(f).Decode(cfg) + if err != nil { + return + } + err = Validate(cfg) return } +func Validate(c *Config) error { + if len(c.VirtualIPs) == 0 { + return errors.New("missing virtualIPs config") + } + raft := c.Cluster.Raft + if raft != nil { + if raft.Dir == "" { + raft.Dir = RaftDir + } + if raft.Level == "" { + raft.Level = "off" + } + if raft.ID == "" { + return errors.New("missing raft.id config") + } + if raft.Address == "" { + return errors.New("missing raft.address config") + } + if len(raft.Peers) < 2 { + return errors.New("minimum number of remote peers: 2") + } + for i, v := range raft.Peers { + if v.ID == "" { + return fmt.Errorf("missing raft.peers[%d].id config", i) + } + if v.Address == "" { + return fmt.Errorf("missing raft.peers[%d].address config", i) + } + } + } + etcd := c.Cluster.Etcd + if etcd != nil { + + } + if raft == nil && etcd == nil { + return errors.New("missing cluster config: raft or etcd") + } + if raft != nil && etcd != nil { + return errors.New("only one cluster config allowed: raft or etcd") + } + return nil +} + type Config struct { - RaftDir string `yaml:"raftDir"` + VirtualIPs []string `yaml:"virtualIPs"` - Leader bool `yaml:"leader"` - - AddressRange []string `yaml:"addressRange"` - - // LocalPeer is the configuration of this host - LocalPeer RaftPeer `yaml:"localPeer"` - - // Peers are all of the peers within the RAFT cluster - RemotePeers []RaftPeer `yaml:"remotePeers"` - - // Interface is the network interface to bind to (default: First Adapter) Interface string `yaml:"interface,omitempty"` + + IPLabel string `yaml:"label,omitempty"` + + LeaderHook string `yaml:"leaderHook,omitempty"` + + FollowerHook string `yaml:"followerHook,omitempty"` + + Cluster Cluster `yaml:"cluster"` +} + +type Cluster struct { + Raft *Raft `yaml:"raft,omitempty"` + Etcd *Etcd `yaml:"etcd,omitempty"` +} + +type Raft struct { + Dir string `yaml:"dir,omitempty"` + ID string `yaml:"id"` + Address string `yaml:"address"` + Level string `yaml:"level"` + Peers []RaftPeer `yaml:"peers"` + TLS *TLS `yaml:"tls,omitempty"` +} + +type TLS struct { + Cert string `yaml:"cert"` + Key string `yaml:"key"` + CA string `yaml:"ca,omitempty"` + Insecure bool `yaml:"insecure,omitempty"` } // RaftPeer details the configuration of all cluster peers type RaftPeer struct { - ID string `yaml:"id"` - - // IP Address of a peer instance + ID string `yaml:"id"` Address string `yaml:"address"` - - // Listening port of this peer instance - Port int `yaml:"port"` +} + +type Etcd struct { + Endpoints []string `yaml:"endpoints"` + Prefix string `yaml:"prefix"` + ClusterName string `yaml:"clusterName"` } diff --git a/pkg/raft/cluster.go b/pkg/raft/cluster.go new file mode 100644 index 0000000..1910240 --- /dev/null +++ b/pkg/raft/cluster.go @@ -0,0 +1,36 @@ +package raft + +import ( + "errors" + + "git.giftfish.de/ston1th/vipman/pkg/cluster" + "github.com/hashicorp/raft" +) + +type Cluster struct { + kv *KV + r *raft.Raft + + stepdown chan struct{} + stop chan struct{} + done chan struct{} + Callbacks cluster.Callbacks +} + +func NewCluster(callbacks cluster.Callbacks) (cluster.Cluster, error) { + if callbacks.Leader == nil { + return nil, errors.New("Leader is nil") + } + if callbacks.Follower == nil { + return nil, errors.New("Follower is nil") + } + return &Cluster{ + kv: &KV{ + m: make(map[string][]byte), + }, + stepdown: make(chan struct{}, 1), + stop: make(chan struct{}, 1), + done: make(chan struct{}, 1), + Callbacks: callbacks, + }, nil +} diff --git a/pkg/cluster/fsm.go b/pkg/raft/fsm.go similarity index 85% rename from pkg/cluster/fsm.go rename to pkg/raft/fsm.go index 21510ba..5a3de37 100644 --- a/pkg/cluster/fsm.go +++ b/pkg/raft/fsm.go @@ -1,11 +1,11 @@ -package cluster +package raft import ( - gogob "encoding/gob" "fmt" "io" "time" + "git.giftfish.de/ston1th/raftbbolt/msgpack" "github.com/hashicorp/raft" ) @@ -31,7 +31,7 @@ type fsm KV func (f *fsm) Apply(l *raft.Log) interface{} { var c raftCmd - if err := gob.Unmarshal(l.Data, &c); err != nil { + if err := msgpack.Unmarshal(l.Data, &c); err != nil { panic(fmt.Sprintf("failed to unmarshal command: %s", err.Error())) } @@ -58,7 +58,7 @@ func (f *fsm) Snapshot() (raft.FSMSnapshot, error) { func (f *fsm) Restore(rc io.ReadCloser) error { o := make(kv) - if err := gogob.NewDecoder(rc).Decode(&o); err != nil { + if err := msgpack.NewDecoder(rc).Decode(&o); err != nil { return err } f.m = o @@ -83,7 +83,7 @@ type fsmSnapshot struct { func (f *fsmSnapshot) Persist(sink raft.SnapshotSink) error { err := func() error { - err := gogob.NewEncoder(sink).Encode(f.m) + err := msgpack.NewEncoder(sink).Encode(f.m) if err != nil { return err } diff --git a/pkg/cluster/kv.go b/pkg/raft/kv.go similarity index 86% rename from pkg/cluster/kv.go rename to pkg/raft/kv.go index 408abce..69ee90f 100644 --- a/pkg/cluster/kv.go +++ b/pkg/raft/kv.go @@ -1,9 +1,10 @@ -package cluster +package raft import ( "errors" "sync" + "git.giftfish.de/ston1th/raftbbolt/msgpack" "github.com/hashicorp/raft" ) @@ -29,7 +30,7 @@ func (kv *KV) Set(k string, v []byte) error { } c := &raftCmd{SET, k, v} - b, err := gob.Marshal(c) + b, err := msgpack.Marshal(c) if err != nil { return err } @@ -45,7 +46,7 @@ func (kv *KV) Delete(k string) error { } c := &raftCmd{Op: DELETE, K: k} - b, err := gob.Marshal(c) + b, err := msgpack.Marshal(c) if err != nil { return err } diff --git a/pkg/cluster/raft.go b/pkg/raft/raft.go similarity index 57% rename from pkg/cluster/raft.go rename to pkg/raft/raft.go index 4e855f7..ec5c648 100644 --- a/pkg/cluster/raft.go +++ b/pkg/raft/raft.go @@ -1,9 +1,8 @@ -package cluster +package raft import ( "context" "fmt" - "io/ioutil" "net" "net/http" "os" @@ -11,11 +10,13 @@ import ( "runtime" "time" + "git.giftfish.de/ston1th/raftbbolt" "git.giftfish.de/ston1th/vipman/pkg/config" + "git.giftfish.de/ston1th/vipman/pkg/raft/tls" "github.com/go-logr/logr" + hclog "github.com/hashicorp/go-hclog" "github.com/hashicorp/raft" - raftboltdb "github.com/hashicorp/raft-boltdb" - "k8s.io/klog" + "k8s.io/klog/v2" ) type raftLog string @@ -24,42 +25,50 @@ func (r *raftLog) Write(p []byte) (n int, err error) { return os.Stdout.Write(append([]byte(*r), p...)) } -var _ = ioutil.Discard +func (c *Cluster) Start(log logr.Logger, raftcfg *config.Config) error { + cfg := raftcfg.Cluster.Raft + hcl := hclog.New(&hclog.LoggerOptions{ + Name: "raft", + Level: hclog.LevelFromString(cfg.Level), + Output: hclog.DefaultOutput, + TimeFormat: "I0102 15:04:05.000000", + }) -// StartRaftCluster - Begins a running instance of the Raft cluster -func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { - - // Create local configuration address - localAddress := fmt.Sprintf("%s:%d", cfg.LocalPeer.Address, cfg.LocalPeer.Port) - - // Begin the Raft configuration rc := raft.DefaultConfig() rc.ProtocolVersion = 3 - rc.LocalID = raft.ServerID(cfg.LocalPeer.ID) - //rl := raftLog(cfg.LocalPeer.ID + " ") - //rc.LogOutput = &rl - rc.LogOutput = ioutil.Discard + rc.LocalID = raft.ServerID(cfg.ID) + rc.Logger = hcl - // 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, os.Stdout) - if err != nil { - return err + pool := 3 + timeout := 10 * time.Second + var transport *raft.NetworkTransport + if cfg.TLS != nil { + log.Info("transport: tls") + s, err := tls.NewStream(cfg) + if err != nil { + return err + } + transport = raft.NewNetworkTransportWithLogger(s, pool, timeout, hcl) + } else { + log.Info("transport: tcp") + address, err := net.ResolveTCPAddr("tcp", cfg.Address) + if err != nil { + return err + } + transport, err = raft.NewTCPTransportWithLogger(cfg.Address, address, pool, timeout, hcl) + if err != nil { + return err + } } // Create Raft structures //snapshots := raft.NewInmemSnapshotStore() - snapshots, err := raft.NewFileSnapshotStore(cfg.RaftDir, retainSnapshotCount, os.Stderr) + snapshots, err := raft.NewFileSnapshotStoreWithLogger(cfg.Dir, retainSnapshotCount, hcl) if err != nil { return fmt.Errorf("file snapshot store: %s", err) } - store := filepath.Join(cfg.RaftDir, "store.db") + store := filepath.Join(cfg.Dir, "store.db") //logStore := raft.NewInmemStore() bootstrap := true @@ -67,7 +76,7 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { bootstrap = false } - stableStore, err := raftboltdb.NewBoltStore(store) + stableStore, err := raftbbolt.NewStore(store) if err != nil { return err } @@ -78,33 +87,14 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { // Add Local Peer configuration.Servers = append(configuration.Servers, raft.Server{ - Suffrage: raft.Voter, - ID: raft.ServerID(cfg.LocalPeer.ID), - Address: raft.ServerAddress(localAddress), + ID: raft.ServerID(cfg.ID), + Address: raft.ServerAddress(cfg.Address), }) - - // 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 !cfg.Leader { - // for _, p := range cfg.RemotePeers { - // peerAddress := fmt.Sprintf("%s:%d", p.Address, p.Port) - // if localAddress != peerAddress { - // configuration.Servers = append(configuration.Servers, raft.Server{ - // ID: raft.ServerID(p.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") - //} - for _, p := range cfg.RemotePeers { - peerAddress := fmt.Sprintf("%s:%d", p.Address, p.Port) - if localAddress != peerAddress { + for _, p := range cfg.Peers { + if cfg.Address != p.Address { configuration.Servers = append(configuration.Servers, raft.Server{ - Suffrage: raft.Voter, - ID: raft.ServerID(p.ID), - Address: raft.ServerAddress(peerAddress)}) + ID: raft.ServerID(p.ID), + Address: raft.ServerAddress(p.Address)}) } } @@ -113,11 +103,6 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { return err } } - //else { - // if err := raft.RecoverCluster(rc, c.stateMachine, logStore, stableStore, snapshots, transport, configuration); err != nil { - // return err - // } - //} // Create RAFT instance c.r, err = raft.NewRaft(rc, (*fsm)(c.kv), logStore, stableStore, snapshots, transport) @@ -126,7 +111,7 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { } c.kv.r = c.r - ticker := time.NewTicker(time.Second) + t := time.NewTicker(time.Second) time.Sleep(time.Second * 3) ctx, cancel := context.WithCancel(context.Background()) @@ -134,7 +119,7 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { leading := false following := false for { - if !leading && localAddress == string(c.r.Leader()) { + if !leading && cfg.Address == string(c.r.Leader()) { leading = true go c.leader(ctx) } else if !following && !leading { @@ -152,9 +137,6 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { } if !leading { leading = true - //cf := c.r.GetConfiguration() - //cf.Error() - //log.Info(fmt.Sprintf("[%s] %#v", cfg.LocalPeer.ID, cf.Configuration())) go c.leader(ctx) } } else if c.r.State() == raft.Follower { @@ -165,20 +147,23 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { } if !following { following = true - //cf := c.r.GetConfiguration() - //cf.Error() - //log.Info(fmt.Sprintf("[%s] %#v", cfg.LocalPeer.ID, cf.Configuration())) go c.follower(ctx) } } - case <-ticker.C: + case <-c.stepdown: + if leading { + cancel() + c.r.LeadershipTransfer().Error() + } case <-c.stop: + t.Stop() cancel() if leading { c.r.LeadershipTransfer().Error() } close(c.done) return nil + case <-t.C: } } return nil @@ -187,7 +172,6 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { func (c *Cluster) leader(ctx context.Context) { defer func() { handleCrash() - //c.Callbacks.OnStoppedLeading() }() c.Callbacks.Leader(ctx, c.kv) } @@ -195,7 +179,6 @@ func (c *Cluster) leader(ctx context.Context) { func (c *Cluster) follower(ctx context.Context) { defer func() { handleCrash() - //c.Callbacks.OnStoppedLeading() }() c.Callbacks.Follower(ctx, c.kv) } @@ -233,6 +216,10 @@ func logPanic(r interface{}) { } } +func (c *Cluster) Stepdown() { + c.stepdown <- struct{}{} +} + func (c *Cluster) Stop() { close(c.stop) <-c.done diff --git a/pkg/raft/tls/stream.go b/pkg/raft/tls/stream.go new file mode 100644 index 0000000..3227cfc --- /dev/null +++ b/pkg/raft/tls/stream.go @@ -0,0 +1,89 @@ +package tls + +import ( + "crypto/tls" + "crypto/x509" + "io/ioutil" + "net" + "os" + "time" + + "git.giftfish.de/ston1th/vipman/pkg/config" + "github.com/hashicorp/raft" +) + +var ( + ciphers = []uint16{ + tls.TLS_CHACHA20_POLY1305_SHA256, + } + curves = []tls.CurveID{ + tls.X25519, + } +) + +func loadCertPool(file string) (p *x509.CertPool, err error) { + if file == "" { + return + } + f, err := os.Open(file) + if err != nil { + return + } + defer f.Close() + buf, err := ioutil.ReadAll(f) + if err != nil { + return + } + certs, err := x509.ParseCertificates(buf) + if err != nil { + return + } + p = x509.NewCertPool() + for _, c := range certs { + p.AddCert(c) + } + return +} + +type stream struct { + net.Listener + insecure bool + ca *x509.CertPool +} + +func NewStream(cfg *config.Raft) (s *stream, err error) { + cert, err := tls.LoadX509KeyPair(cfg.TLS.Cert, cfg.TLS.Key) + if err != nil { + return + } + ca, err := loadCertPool(cfg.TLS.CA) + if err != nil { + return + } + l, err := tls.Listen("tcp", cfg.Address, &tls.Config{ + MinVersion: tls.VersionTLS13, + CipherSuites: ciphers, + CurvePreferences: curves, + PreferServerCipherSuites: true, + Certificates: []tls.Certificate{cert}, + }) + if err != nil { + return + } + s = &stream{l, cfg.TLS.Insecure, ca} + return +} + +func (s *stream) Dial(address raft.ServerAddress, timeout time.Duration) (net.Conn, error) { + return tls.DialWithDialer( + &net.Dialer{Timeout: timeout}, + "tcp", string(address), + &tls.Config{ + MinVersion: tls.VersionTLS13, + CipherSuites: ciphers, + CurvePreferences: curves, + InsecureSkipVerify: s.insecure, + RootCAs: s.ca, + }, + ) +} diff --git a/pkg/vip/arp.go b/pkg/vip/arp.go index 5d73258..4333498 100644 --- a/pkg/vip/arp.go +++ b/pkg/vip/arp.go @@ -5,6 +5,6 @@ package vip import "fmt" // ARPSendGratuitous is only supported on Linux, so return an error -func ARPSendGratuitous(address, ifaceName string) error { +func ARPSendGratuitous(cidr, iface string) error { return fmt.Errorf("Unsupported on this OS") } diff --git a/pkg/vip/arp_linux.go b/pkg/vip/arp_linux.go index a8b1642..cd08399 100644 --- a/pkg/vip/arp_linux.go +++ b/pkg/vip/arp_linux.go @@ -11,8 +11,6 @@ import ( "net" "syscall" "unsafe" - - log "github.com/sirupsen/logrus" ) const ( @@ -156,21 +154,20 @@ func sendARP(iface *net.Interface, m *arpMessage) error { } // ARPSendGratuitous sends a gratuitous ARP message via the specified interface. -func ARPSendGratuitous(address, ifaceName string) error { - iface, err := net.InterfaceByName(ifaceName) +func ARPSendGratuitous(cidr, iface string) error { + i, err := net.InterfaceByName(iface) if err != nil { - return fmt.Errorf("failed to get interface %q: %v", ifaceName, err) + return fmt.Errorf("failed to get interface %q: %v", iface, err) + } + ip, _, err := net.ParseCIDR(cidr) + if err != nil { + return fmt.Errorf("failed to parse cidr: %s", cidr) } - ip := net.ParseIP(address) - if ip == nil { - return fmt.Errorf("failed to parse address %s", ip) - } - - log.Infof("Broadcasting ARP update for %s (%s) via %s", address, iface.HardwareAddr, iface.Name) - m, err := gratuitousARP(ip, iface.HardwareAddr) + //log.Infof("Broadcasting ARP update for %s (%s) via %s", address, iface.HardwareAddr, iface.Name) + m, err := gratuitousARP(ip, i.HardwareAddr) if err != nil { return err } - return sendARP(iface, m) + return sendARP(i, m) } diff --git a/pkg/vip/vip.go b/pkg/vip/vip.go index f4a716e..27dab89 100644 --- a/pkg/vip/vip.go +++ b/pkg/vip/vip.go @@ -1,142 +1,97 @@ package vip import ( - "sync" - - "github.com/pkg/errors" + "errors" "github.com/vishvananda/netlink" + "net" ) -const ( - defaultValidLft = 60 -) +var ErrNoDefaultInterface = errors.New("no default interface found") -// Network is an interface that enable managing operations for a given IP -type Network interface { - AddIP() error - DeleteIP() error - IsSet() (bool, error) - IP() string - SetIP(ip string) error - Interface() string -} - -// network - This allows network configuration -type network struct { - mu sync.Mutex - - address *netlink.Addr - link netlink.Link - isDNS bool -} - -// NewConfig will attempt to provide an interface to the kernel network configuration -func NewConfig(address string, iface string) (Network, error) { - result := &network{} - - link, err := netlink.LinkByName(iface) +func DefaultInterface() (iface string, err error) { + routes, err := netlink.RouteGet(net.IPv4bcast) if err != nil { - return result, errors.Wrapf(err, "could not get link for interface '%s'", iface) - } - result.link = link - - if IsIP(address) { - result.address, err = netlink.ParseAddr(address + "/32") - if err != nil { - return result, errors.Wrapf(err, "could not parse address '%s'", address) - } - return result, nil - } - - // try to resolve the address - ip, err := lookupHost(address) - result.isDNS = err == nil - if err != nil { - return nil, err - } - - // we're able to resolve store this as the initial IP - if result.address, err = netlink.ParseAddr(ip + "/32"); err != nil { - return result, err - } - // set ValidLft so that the VIP expires if the DNS entry is updated, otherwise it'll be refreshed by the DNS prober - result.address.ValidLft = defaultValidLft - - return result, err -} - -//AddIP - Add an IP address to the interface -func (configurator *network) AddIP() error { - if err := netlink.AddrReplace(configurator.link, configurator.address); err != nil { - return errors.Wrap(err, "could not add ip") - } - return nil -} - -//DeleteIP - Remove an IP address from the interface -func (configurator *network) DeleteIP() error { - result, err := configurator.IsSet() - if err != nil { - return errors.Wrap(err, "ip check in DeleteIP failed") - } - - // Nothing to delete - if !result { - return nil - } - - if err = netlink.AddrDel(configurator.link, configurator.address); err != nil { - return errors.Wrap(err, "could not delete ip") - } - - return nil -} - -// IsSet - Check to see if VIP is set -func (configurator *network) IsSet() (result bool, err error) { - var addresses []netlink.Addr - - addresses, err = netlink.AddrList(configurator.link, 0) - if err != nil { - err = errors.Wrap(err, "could not list addresses") - return } - - for _, address := range addresses { - if address.Equal(*configurator.address) { - return true, nil + if len(routes) >= 1 { + link, e := netlink.LinkByIndex(routes[0].LinkIndex) + if e != nil { + err = e + return } + return link.Attrs().Name, nil } - - return false, nil + err = ErrNoDefaultInterface + return } -// SetIP updates the IP that is used -func (configurator *network) SetIP(ip string) error { - configurator.mu.Lock() - defer configurator.mu.Unlock() +type Network struct { + iface string + label string + link netlink.Link +} - addr, err := netlink.ParseAddr(ip + "/32") +func NewNetwork(iface string) (n *Network, err error) { + return NewNetworkWithLabel(iface, "") +} + +func NewNetworkWithLabel(iface, label string) (n *Network, err error) { + if iface == "" { + iface, err = DefaultInterface() + if err != nil { + return + } + } + link, err := netlink.LinkByName(iface) + if err != nil { + return + } + n = &Network{iface, label, link} + return +} + +func (n *Network) Interface() string { + return n.iface +} + +func (n *Network) HasIP(a *netlink.Addr) bool { + addrs, err := netlink.AddrList(n.link, 0) + if err != nil { + return false + } + for _, addr := range addrs { + if addr.Equal(*a) { + return true + } + } + return false +} + +func (n *Network) AddIP(cidr string) error { + addr, err := netlink.ParseAddr(cidr) if err != nil { return err } - if configurator.address != nil && configurator.isDNS { - addr.ValidLft = defaultValidLft + if n.HasIP(addr) { + return nil } - configurator.address = addr - return nil + if n.label != "" { + addr.Label = n.iface + ":" + n.label + } + err = netlink.AddrReplace(n.link, addr) + if err != nil { + return err + } + return ARPSendGratuitous(cidr, n.iface) } -// IP - return the IP Address -func (configurator *network) IP() string { - configurator.mu.Lock() - defer configurator.mu.Unlock() - - return configurator.address.IP.String() -} - -// Interface - return the Interface name -func (configurator *network) Interface() string { - return configurator.link.Attrs().Name +func (n *Network) DeleteIP(cidr string) error { + addr, err := netlink.ParseAddr(cidr) + if err != nil { + return err + } + if !n.HasIP(addr) { + return nil + } + return netlink.AddrDel(n.link, addr) } diff --git a/vipman.yaml b/vipman.yaml index 1307ae8..34ea273 100644 --- a/vipman.yaml +++ b/vipman.yaml @@ -1,14 +1,29 @@ -raftDir: "" -addressRange: +virtualIPs: - "" -localPeer: - id: "" - address: "" - port: 0 -remotePeers: -- id: "" - address: "" - port: 0 -- id: "" - address: "" - port: 0 +interface: "" +label: "" +leaderHook: "" +followerHook: "" +cluster: + raft: + dir: "" + id: "" + address: "" + level: "" + peers: + - id: "" + address: "" + - id: "" + address: "" + tls: + cert: "" + key: "" + ca: "" + insecure: true + etcd: + endpoints: + - "" + - "" + - "" + prefix: "" + clusterName: ""