diff --git a/cmd/vipman/main.go b/cmd/vipman/main.go index 1f8dc6f..678d150 100644 --- a/cmd/vipman/main.go +++ b/cmd/vipman/main.go @@ -26,23 +26,41 @@ func main() { if err != nil { klog.Fatalf("Initialization failed: %s", err) } - c, err := cluster.InitCluster(cluster.LeaderCallbacks{ - OnStartedLeading: func(ctx context.Context) { + c, err := cluster.InitCluster(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: } - klog.Infof("[%s] leading", cfg.LocalPeer.ID) } }, - OnStoppedLeading: func() { - klog.Infof("[%s] stopped leading", cfg.LocalPeer.ID) - }, - OnNewLeader: func(id string) { - klog.Infof("[%s] new leader: %s", cfg.LocalPeer.ID, id) + 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: + } + } }, }) if err != nil { diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index 637e2b6..9e58335 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -3,39 +3,37 @@ package cluster import ( "context" "errors" + + "github.com/hashicorp/raft" ) // Cluster - The Cluster object manages the state of the cluster for a particular node type Cluster struct { - stateMachine FSM - stop chan struct{} - done chan struct{} - Callbacks LeaderCallbacks + kv *KV + r *raft.Raft + + stop chan struct{} + done chan struct{} + Callbacks Callbacks } -type LeaderCallbacks struct { - // OnStartedLeading is called when a LeaderElector client starts leading - OnStartedLeading func(context.Context) - // OnStoppedLeading is called when a LeaderElector client stops leading - OnStoppedLeading func() - // OnNewLeader is called when the client observes a leader that is - // not the previously observed leader. This includes the first observed - // leader when the client starts. - OnNewLeader func(string) +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 LeaderCallbacks) (*Cluster, error) { - if callbacks.OnStartedLeading == nil { - return nil, errors.New("OnStartedLeading is nil") +func InitCluster(callbacks Callbacks) (*Cluster, error) { + if callbacks.Leader == nil { + return nil, errors.New("Leader is nil") } - if callbacks.OnStoppedLeading == nil { - return nil, errors.New("OnStoppedLeading is nil") - } - if callbacks.OnNewLeader == nil { - return nil, errors.New("OnNewLeader 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, diff --git a/pkg/cluster/fsm.go b/pkg/cluster/fsm.go new file mode 100644 index 0000000..21510ba --- /dev/null +++ b/pkg/cluster/fsm.go @@ -0,0 +1,99 @@ +package cluster + +import ( + gogob "encoding/gob" + "fmt" + "io" + "time" + + "github.com/hashicorp/raft" +) + +const ( + retainSnapshotCount = 2 + raftTimeout = 10 * time.Second +) + +type Op uint8 + +const ( + SET Op = iota + DELETE +) + +type raftCmd struct { + Op Op + K string + V []byte +} + +type fsm KV + +func (f *fsm) Apply(l *raft.Log) interface{} { + var c raftCmd + if err := gob.Unmarshal(l.Data, &c); err != nil { + panic(fmt.Sprintf("failed to unmarshal command: %s", err.Error())) + } + + switch c.Op { + case SET: + f.applySet(c.K, c.V) + case DELETE: + f.applyDelete(c.K) + default: + panic(fmt.Sprintf("unrecognized command op: %s", c.Op)) + } + return nil +} + +func (f *fsm) Snapshot() (raft.FSMSnapshot, error) { + f.mu.Lock() + defer f.mu.Unlock() + m := make(kv) + for k, v := range f.m { + m[k] = v + } + return &fsmSnapshot{m: m}, nil +} + +func (f *fsm) Restore(rc io.ReadCloser) error { + o := make(kv) + if err := gogob.NewDecoder(rc).Decode(&o); err != nil { + return err + } + f.m = o + return nil +} + +func (f *fsm) applySet(k string, v []byte) { + f.mu.Lock() + f.m[k] = v + f.mu.Unlock() +} + +func (f *fsm) applyDelete(k string) { + f.mu.Lock() + delete(f.m, k) + f.mu.Unlock() +} + +type fsmSnapshot struct { + m kv +} + +func (f *fsmSnapshot) Persist(sink raft.SnapshotSink) error { + err := func() error { + err := gogob.NewEncoder(sink).Encode(f.m) + if err != nil { + return err + } + return sink.Close() + }() + + if err != nil { + sink.Cancel() + } + return err +} + +func (f *fsmSnapshot) Release() {} diff --git a/pkg/cluster/kv.go b/pkg/cluster/kv.go new file mode 100644 index 0000000..408abce --- /dev/null +++ b/pkg/cluster/kv.go @@ -0,0 +1,55 @@ +package cluster + +import ( + "errors" + "sync" + + "github.com/hashicorp/raft" +) + +var ErrNotLeader = errors.New("not leader") + +type kv map[string][]byte + +type KV struct { + mu sync.Mutex + m kv + r *raft.Raft +} + +func (kv *KV) Get(k string) []byte { + kv.mu.Lock() + defer kv.mu.Unlock() + return kv.m[k] +} + +func (kv *KV) Set(k string, v []byte) error { + if kv.r.State() != raft.Leader { + return ErrNotLeader + } + + c := &raftCmd{SET, k, v} + b, err := gob.Marshal(c) + if err != nil { + return err + } + + f := kv.r.Apply(b, raftTimeout) + return f.Error() +} + +// Delete deletes the given key. +func (kv *KV) Delete(k string) error { + if kv.r.State() != raft.Leader { + return ErrNotLeader + } + + c := &raftCmd{Op: DELETE, K: k} + b, err := gob.Marshal(c) + if err != nil { + return err + } + + f := kv.r.Apply(b, raftTimeout) + return f.Error() +} diff --git a/pkg/cluster/marshaler.go b/pkg/cluster/marshaler.go new file mode 100644 index 0000000..ce625bf --- /dev/null +++ b/pkg/cluster/marshaler.go @@ -0,0 +1,21 @@ +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/cluster/raft.go b/pkg/cluster/raft.go index a805512..4e855f7 100644 --- a/pkg/cluster/raft.go +++ b/pkg/cluster/raft.go @@ -7,6 +7,7 @@ import ( "net" "net/http" "os" + "path/filepath" "runtime" "time" @@ -33,10 +34,11 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { // 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 + //rl := raftLog(cfg.LocalPeer.ID + " ") + //rc.LogOutput = &rl + rc.LogOutput = ioutil.Discard // Initialize communication address, err := net.ResolveTCPAddr("tcp", localAddress) @@ -51,44 +53,59 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { } // Create Raft structures - snapshots := raft.NewInmemSnapshotStore() + //snapshots := raft.NewInmemSnapshotStore() + snapshots, err := raft.NewFileSnapshotStore(cfg.RaftDir, retainSnapshotCount, os.Stderr) + if err != nil { + return fmt.Errorf("file snapshot store: %s", err) + } + + store := filepath.Join(cfg.RaftDir, "store.db") + //logStore := raft.NewInmemStore() bootstrap := true - if _, err := os.Stat(cfg.DBFile); err == nil { + if _, err := os.Stat(store); err == nil { bootstrap = false } - stableStore, err := raftboltdb.NewBoltStore(cfg.DBFile) - if err != nil { - return err - } - logStore, err := raftboltdb.NewBoltStore(cfg.DBFile + ".log") + + stableStore, err := raftboltdb.NewBoltStore(store) if err != nil { return err } + logStore := stableStore // Cluster configuration configuration := raft.Configuration{} // Add Local Peer configuration.Servers = append(configuration.Servers, raft.Server{ - ID: raft.ServerID(cfg.LocalPeer.ID), - Address: raft.ServerAddress(localAddress), + Suffrage: raft.Voter, + ID: raft.ServerID(cfg.LocalPeer.ID), + Address: raft.ServerAddress(localAddress), }) // 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)}) - } + //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 { + configuration.Servers = append(configuration.Servers, raft.Server{ + Suffrage: raft.Voter, + 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") } if bootstrap { @@ -103,48 +120,62 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { //} // Create RAFT instance - raftServer, err := raft.NewRaft(rc, c.stateMachine, logStore, stableStore, snapshots, transport) + c.r, err = raft.NewRaft(rc, (*fsm)(c.kv), logStore, stableStore, snapshots, transport) if err != nil { return err } + c.kv.r = c.r ticker := time.NewTicker(time.Second) - time.Sleep(time.Second * 5) + time.Sleep(time.Second * 3) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - lead := raftServer.Leader() leading := false + following := false for { - l := raftServer.Leader() - if lead != l && string(l) != "" { - go c.Callbacks.OnNewLeader(string(l)) - lead = l - } - if !leading && localAddress == string(raftServer.Leader()) { + if !leading && localAddress == string(c.r.Leader()) { leading = true - go c.Run(ctx) + go c.leader(ctx) + } else if !following && !leading { + following = true + go c.follower(ctx) } select { - case leader := <-raftServer.LeaderCh(): + case leader := <-c.r.LeaderCh(): if leader { + if following { + cancel() + following = false + ctx, cancel = context.WithCancel(context.Background()) + } if !leading { leading = true - go c.Run(ctx) + //cf := c.r.GetConfiguration() + //cf.Error() + //log.Info(fmt.Sprintf("[%s] %#v", cfg.LocalPeer.ID, cf.Configuration())) + go c.leader(ctx) } - } else { + } else if c.r.State() == raft.Follower { if leading { cancel() leading = false ctx, cancel = context.WithCancel(context.Background()) } + 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.stop: + cancel() if leading { - cancel() - raftServer.LeadershipTransfer() + c.r.LeadershipTransfer().Error() } close(c.done) return nil @@ -153,12 +184,20 @@ func (c *Cluster) StartRaftCluster(log logr.Logger, cfg *config.Config) error { return nil } -func (c *Cluster) Run(ctx context.Context) { +func (c *Cluster) leader(ctx context.Context) { defer func() { handleCrash() - c.Callbacks.OnStoppedLeading() + //c.Callbacks.OnStoppedLeading() }() - c.Callbacks.OnStartedLeading(ctx) + c.Callbacks.Leader(ctx, c.kv) +} + +func (c *Cluster) follower(ctx context.Context) { + defer func() { + handleCrash() + //c.Callbacks.OnStoppedLeading() + }() + c.Callbacks.Follower(ctx, c.kv) } var reallyCrash = false @@ -197,4 +236,13 @@ func logPanic(r interface{}) { func (c *Cluster) Stop() { close(c.stop) <-c.done + s := make(chan struct{}) + go func() { + c.r.Shutdown().Error() + close(s) + }() + select { + case <-s: + case <-time.After(time.Second * 5): + } } diff --git a/pkg/cluster/state.go b/pkg/cluster/state.go deleted file mode 100644 index 02f12f6..0000000 --- a/pkg/cluster/state.go +++ /dev/null @@ -1,32 +0,0 @@ -package cluster - -import ( - "io" - - "github.com/hashicorp/raft" -) - -type FSM struct { -} - -func (fsm FSM) Apply(log *raft.Log) interface{} { - return nil -} - -func (fsm FSM) Restore(snap io.ReadCloser) error { - return nil -} - -func (fsm FSM) Snapshot() (raft.FSMSnapshot, error) { - return Snapshot{}, nil -} - -type Snapshot struct { -} - -func (snapshot Snapshot) Persist(sink raft.SnapshotSink) error { - return nil -} - -func (snapshot Snapshot) Release() { -} diff --git a/pkg/config/config.go b/pkg/config/config.go index 15e497f..791ef7f 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -20,7 +20,7 @@ func ParseFile(file string) (cfg *Config, err error) { } type Config struct { - DBFile string `yaml:"dbFile"` + RaftDir string `yaml:"raftDir"` Leader bool `yaml:"leader"` diff --git a/run.sh b/run.sh index aa352d6..6343783 100755 --- a/run.sh +++ b/run.sh @@ -1,7 +1,7 @@ #!/bin/sh cd data -#rm -f node* +rm -rf node* ../vipman -config config1.yaml & ../vipman -config config2.yaml & ../vipman -config config3.yaml diff --git a/vipman.yaml b/vipman.yaml index 9de2435..1307ae8 100644 --- a/vipman.yaml +++ b/vipman.yaml @@ -1,4 +1,4 @@ -dbFile: "" +raftDir: "" addressRange: - "" localPeer: