wesher/cluster.go

264 lines
6.2 KiB
Go
Raw Normal View History

2019-03-25 01:02:10 +01:00
package main
import (
"bytes"
"crypto/rand"
"encoding/base64"
"encoding/gob"
"encoding/json"
"fmt"
"io/ioutil"
"net"
"os"
"path"
"time"
"github.com/hashicorp/memberlist"
2020-01-31 19:31:50 +01:00
"github.com/mattn/go-isatty"
2020-04-19 13:37:49 +02:00
"github.com/pkg/errors"
2019-03-25 01:02:10 +01:00
"github.com/sirupsen/logrus"
2020-04-19 13:37:49 +02:00
"go.uber.org/multierr"
2019-03-25 01:02:10 +01:00
)
// ClusterState keeps track of information needed to rejoin the cluster
type ClusterState struct {
ClusterKey []byte
Nodes []node
}
type cluster struct {
localName string // used to avoid LocalNode(); should not change
ml *memberlist.Memberlist
wg *wgState
state *ClusterState
events chan memberlist.NodeEvent
}
const statePath = "/var/lib/wesher/state.json"
func newCluster(config *config, wg *wgState) (*cluster, error) {
clusterKey := config.ClusterKey
state := &ClusterState{}
if !config.Init {
loadState(state)
}
2019-03-25 01:02:10 +01:00
if len(clusterKey) == 0 {
clusterKey = state.ClusterKey
}
if len(clusterKey) == 0 {
clusterKey = make([]byte, clusterKeyLen)
_, err := rand.Read(clusterKey)
if err != nil {
return nil, err
}
// TODO: refactor this into subcommand ("showkey"?)
if isatty.IsTerminal(os.Stdout.Fd()) {
fmt.Printf("new cluster key generated: %s\n", base64.StdEncoding.EncodeToString(clusterKey))
}
}
state.ClusterKey = clusterKey
// we check for mutual exclusion in config.go
bindAddr := config.BindAddr
if config.BindIface != "" {
iface, err := net.InterfaceByName(config.BindIface)
if err != nil {
return nil, err
}
addrs, err := iface.Addrs()
if err != nil {
return nil, err
}
if len(addrs) > 0 {
if addr, ok := addrs[0].(*net.IPNet); ok {
bindAddr = addr.IP.String()
}
}
}
2019-03-25 01:02:10 +01:00
mlConfig := memberlist.DefaultWANConfig()
mlConfig.LogOutput = logrus.StandardLogger().WriterLevel(logrus.DebugLevel)
mlConfig.SecretKey = clusterKey
mlConfig.BindAddr = bindAddr
2019-03-25 01:02:10 +01:00
mlConfig.BindPort = config.ClusterPort
mlConfig.AdvertisePort = config.ClusterPort
if config.UseIPAsName && config.BindAddr != "0.0.0.0" {
mlConfig.Name = config.BindAddr
}
ml, err := memberlist.Create(mlConfig)
if err != nil {
return nil, err
}
cluster := cluster{
localName: ml.LocalNode().Name,
ml: ml,
wg: wg,
// The big channel buffer is a work-around for https://github.com/hashicorp/memberlist/issues/23
// More than this many simultaneous events will deadlock cluster.members()
events: make(chan memberlist.NodeEvent, 100),
state: state,
2019-03-25 01:02:10 +01:00
}
mlConfig.Conflict = &cluster
mlConfig.Events = &memberlist.ChannelEventDelegate{Ch: cluster.events}
mlConfig.Delegate = &cluster
2019-03-27 21:37:54 +01:00
wg.assignOverlayAddr((*net.IPNet)(config.OverlayNet), cluster.localName)
2019-03-25 01:02:10 +01:00
ml.UpdateNode(1 * time.Second) // we currently do not update after creation
return &cluster, nil
}
func (c *cluster) NotifyConflict(node, other *memberlist.Node) {
logrus.Errorf("node name conflict detected: %s", other.Name)
}
// none if these are used
func (c *cluster) NotifyMsg([]byte) {}
func (c *cluster) GetBroadcasts(overhead, limit int) [][]byte { return nil }
func (c *cluster) LocalState(join bool) []byte { return nil }
func (c *cluster) MergeRemoteState(buf []byte, join bool) {}
type nodeMeta struct {
OverlayAddr net.IPNet
2019-03-25 01:02:10 +01:00
PubKey string
}
func (c *cluster) NodeMeta(limit int) []byte {
buf := &bytes.Buffer{}
if err := gob.NewEncoder(buf).Encode(nodeMeta{
OverlayAddr: c.wg.OverlayAddr,
PubKey: c.wg.PubKey.String(),
2019-03-25 01:02:10 +01:00
}); err != nil {
logrus.Errorf("could not encode local state: %s", err)
return nil
}
if buf.Len() > limit {
logrus.Errorf("could not fit node metadata into %d bytes", limit)
return nil
}
return buf.Bytes()
}
func decodeNodeMeta(b []byte) (nodeMeta, error) {
// TODO: we blindly trust the info we get from the peers; We should be more defensive to limit the damage a leaked
// PSK can cause.
2019-03-25 01:02:10 +01:00
nm := nodeMeta{}
if err := gob.NewDecoder(bytes.NewReader(b)).Decode(&nm); err != nil {
2020-04-19 13:37:49 +02:00
return nm, errors.Wrap(err, "could not decode node meta")
2019-03-25 01:02:10 +01:00
}
return nm, nil
}
func (c *cluster) join(addrs []string) error {
if len(addrs) == 0 {
for _, n := range c.state.Nodes {
addrs = append(addrs, n.Addr.String())
}
}
if _, err := c.ml.Join(addrs); err != nil {
return err
} else if len(addrs) > 0 && c.ml.NumMembers() < 2 {
return errors.New("could not join to any of the provided addresses")
}
return nil
}
func (c *cluster) leave() {
c.saveState()
c.ml.Leave(10 * time.Second)
2020-04-19 13:37:49 +02:00
c.ml.Shutdown() //nolint: errcheck
2019-03-25 01:02:10 +01:00
}
func (c *cluster) members() (<-chan []node, <-chan error) {
changes := make(chan []node)
errc := make(chan error, 1)
go func() {
for {
event := <-c.events
if event.Node.Name == c.localName {
// ignore events about ourselves
continue
}
switch event.Event {
case memberlist.NodeJoin:
logrus.Infof("node %s joined", event.Node)
case memberlist.NodeUpdate:
logrus.Infof("node %s updated", event.Node)
case memberlist.NodeLeave:
logrus.Infof("node %s left", event.Node)
}
nodes := make([]node, 0)
var errs error
for _, n := range c.ml.Members() {
if n.Name == c.localName {
continue
}
meta, err := decodeNodeMeta(n.Meta)
if err != nil {
2020-04-19 13:37:49 +02:00
errs = multierr.Append(errs, err)
2019-03-25 01:02:10 +01:00
continue
}
nodes = append(nodes, node{
2019-03-26 23:26:29 +01:00
Name: n.Name,
2019-03-25 01:02:10 +01:00
Addr: n.Addr,
nodeMeta: meta,
})
}
c.state.Nodes = nodes
changes <- nodes
if errs != nil {
errc <- errs
}
c.saveState()
}
}()
return changes, errc
}
type node struct {
2019-03-26 23:26:29 +01:00
Name string
2019-03-25 01:02:10 +01:00
Addr net.IP
nodeMeta
}
func (n *node) String() string {
return n.Addr.String()
}
func (c *cluster) saveState() error {
if err := os.MkdirAll(path.Dir(statePath), 0700); err != nil {
return err
}
stateOut, err := json.MarshalIndent(c.state, "", " ")
if err != nil {
return err
}
2019-08-24 16:51:30 +02:00
return ioutil.WriteFile(statePath, stateOut, 0600)
2019-03-25 01:02:10 +01:00
}
func loadState(cs *ClusterState) {
2019-03-25 01:02:10 +01:00
content, err := ioutil.ReadFile(statePath)
if err != nil {
if !os.IsNotExist(err) {
logrus.Warnf("could not open state in %s: %s", statePath, err)
}
return
2019-03-25 01:02:10 +01:00
}
// avoid partially unmarshalled content by using a temp var
csTmp := &ClusterState{}
if err := json.Unmarshal(content, csTmp); err != nil {
2019-03-25 01:02:10 +01:00
logrus.Warnf("could not decode state: %s", err)
} else {
*cs = *csTmp
2019-03-25 01:02:10 +01:00
}
}