diff --git a/server/join_command.go b/server/join_command.go index 55359860d..ed262b1bd 100644 --- a/server/join_command.go +++ b/server/join_command.go @@ -37,11 +37,11 @@ func (c *JoinCommand) CommandName() string { } // Join a server to the cluster -func (c *JoinCommand) Apply(server raft.Server) (interface{}, error) { - ps, _ := server.Context().(*PeerServer) +func (c *JoinCommand) Apply(context raft.Context) (interface{}, error) { + ps, _ := context.Server().Context().(*PeerServer) b := make([]byte, 8) - binary.PutUvarint(b, server.CommitIndex()) + binary.PutUvarint(b, context.CommitIndex()) // Make sure we're not getting a cached value from the registry. ps.registry.Invalidate(c.Name) @@ -54,14 +54,14 @@ func (c *JoinCommand) Apply(server raft.Server) (interface{}, error) { // Check peer number in the cluster if ps.registry.Count() == ps.MaxClusterSize { log.Debug("Reject join request from ", c.Name) - return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMorePeer, "", server.CommitIndex()) + return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMorePeer, "", context.CommitIndex()) } // Add to shared peer registry. ps.registry.Register(c.Name, c.RaftURL, c.EtcdURL) // Add peer in raft - err := server.AddPeer(c.Name, "") + err := context.Server().AddPeer(c.Name, "") // Add peer stats if c.Name != ps.RaftServer().Name() { diff --git a/server/peer_server.go b/server/peer_server.go index 1a409fd6b..2948da2a9 100644 --- a/server/peer_server.go +++ b/server/peer_server.go @@ -440,15 +440,15 @@ func (s *PeerServer) raftEventLogger(event raft.Event) { switch event.Type() { case raft.StateChangeEventType: - fmt.Printf("[%s] State changed from '%v' to '%v'.\n", s.name, prevValue, value) + log.Infof("%s: state changed from '%v' to '%v'.", s.name, prevValue, value) case raft.TermChangeEventType: - fmt.Printf("[%s] Term #%v started.\n", s.name, value) + log.Infof("%s: term #%v started.", s.name, value) case raft.LeaderChangeEventType: - fmt.Printf("[%s] Leader changed from '%v' to '%v'.\n", s.name, prevValue, value) + log.Infof("%s: leader changed from '%v' to '%v'.", s.name, prevValue, value) case raft.AddPeerEventType: - fmt.Printf("[%s] Peer added: '%v'\n", s.name, value) + log.Infof("%s: peer added: '%v'", s.name, value) case raft.RemovePeerEventType: - fmt.Printf("[%s] Peer removed: '%v'\n", s.name, value) + log.Infof("%s: peer removed: '%v'", s.name, value) } } diff --git a/server/remove_command.go b/server/remove_command.go index ab1dc2650..2057acc4a 100644 --- a/server/remove_command.go +++ b/server/remove_command.go @@ -23,8 +23,8 @@ func (c *RemoveCommand) CommandName() string { } // Remove a server from the cluster -func (c *RemoveCommand) Apply(server raft.Server) (interface{}, error) { - ps, _ := server.Context().(*PeerServer) +func (c *RemoveCommand) Apply(context raft.Context) (interface{}, error) { + ps, _ := context.Server().Context().(*PeerServer) // Remove node from the shared registry. err := ps.registry.Unregister(c.Name) @@ -38,21 +38,21 @@ func (c *RemoveCommand) Apply(server raft.Server) (interface{}, error) { } // Remove peer in raft - err = server.RemovePeer(c.Name) + err = context.Server().RemovePeer(c.Name) if err != nil { log.Debugf("Unable to remove peer: %s (%v)", c.Name, err) return []byte{0}, err } - if c.Name == server.Name() { + if c.Name == context.Server().Name() { // the removed node is this node // if the node is not replaying the previous logs // and the node has sent out a join request in this // start. It is sure that this node received a new remove // command and need to be removed - if server.CommitIndex() > ps.joinIndex && ps.joinIndex != 0 { - log.Debugf("server [%s] is removed", server.Name()) + if context.CommitIndex() > ps.joinIndex && ps.joinIndex != 0 { + log.Debugf("server [%s] is removed", context.Server().Name()) os.Exit(0) } else { // else ignore remove @@ -61,7 +61,7 @@ func (c *RemoveCommand) Apply(server raft.Server) (interface{}, error) { } b := make([]byte, 8) - binary.PutUvarint(b, server.CommitIndex()) + binary.PutUvarint(b, context.CommitIndex()) return b, err } diff --git a/store/v2/compare_and_swap_command.go b/store/v2/compare_and_swap_command.go index 79deec5e4..1af1b5032 100644 --- a/store/v2/compare_and_swap_command.go +++ b/store/v2/compare_and_swap_command.go @@ -27,8 +27,8 @@ func (c *CompareAndSwapCommand) CommandName() string { } // Set the key-value pair if the current value of the key equals to the given prevValue -func (c *CompareAndSwapCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c *CompareAndSwapCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) e, err := s.CompareAndSwap(c.Key, c.PrevValue, c.PrevIndex, c.Value, c.ExpireTime) diff --git a/store/v2/create_command.go b/store/v2/create_command.go index d38e51ecf..8c5b4a82c 100644 --- a/store/v2/create_command.go +++ b/store/v2/create_command.go @@ -27,8 +27,8 @@ func (c *CreateCommand) CommandName() string { } // Create node -func (c *CreateCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c *CreateCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) e, err := s.Create(c.Key, c.Dir, c.Value, c.Unique, c.ExpireTime) diff --git a/store/v2/delete_command.go b/store/v2/delete_command.go index 12104d094..bbd8ecc70 100644 --- a/store/v2/delete_command.go +++ b/store/v2/delete_command.go @@ -23,8 +23,8 @@ func (c *DeleteCommand) CommandName() string { } // Delete the key -func (c *DeleteCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c *DeleteCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) if c.Recursive { // recursive implies dir diff --git a/store/v2/set_command.go b/store/v2/set_command.go index 9c4e51369..bceb3f192 100644 --- a/store/v2/set_command.go +++ b/store/v2/set_command.go @@ -26,8 +26,8 @@ func (c *SetCommand) CommandName() string { } // Create node -func (c *SetCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c *SetCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) // create a new node or replace the old node. e, err := s.Set(c.Key, c.Dir, c.Value, c.ExpireTime) diff --git a/store/v2/sync_command.go b/store/v2/sync_command.go index adfbcccd4..f51b3eeaf 100644 --- a/store/v2/sync_command.go +++ b/store/v2/sync_command.go @@ -20,8 +20,8 @@ func (c SyncCommand) CommandName() string { return "etcd:sync" } -func (c SyncCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c SyncCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) s.DeleteExpiredKeys(c.Time) return nil, nil diff --git a/store/v2/update_command.go b/store/v2/update_command.go index a677c85a2..a4e7e5990 100644 --- a/store/v2/update_command.go +++ b/store/v2/update_command.go @@ -24,8 +24,8 @@ func (c *UpdateCommand) CommandName() string { } // Create node -func (c *UpdateCommand) Apply(server raft.Server) (interface{}, error) { - s, _ := server.StateMachine().(store.Store) +func (c *UpdateCommand) Apply(context raft.Context) (interface{}, error) { + s, _ := context.Server().StateMachine().(store.Store) e, err := s.Update(c.Key, c.Value, c.ExpireTime) diff --git a/tests/functional/simple_snapshot_test.go b/tests/functional/simple_snapshot_test.go index 29872d648..a93ecb39d 100644 --- a/tests/functional/simple_snapshot_test.go +++ b/tests/functional/simple_snapshot_test.go @@ -89,7 +89,7 @@ func TestSimpleSnapshot(t *testing.T) { index, _ = strconv.Atoi(snapshots[0].Name()[2:6]) - if index <= 1014 || index >= 1017 { + if index <= 1014 || index > 1017 { t.Fatal("wrong name of snapshot :", snapshots[0].Name()) } }