Raft fixes, integrate logging.

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) {
+func (c *JoinCommand) Apply(context raft.Context) (interface{}, error) {
-	ps, _ := server.Context().(*PeerServer)
+	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() {

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)
 	}
 }
 

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) {
+func (c *RemoveCommand) Apply(context raft.Context) (interface{}, error) {
-	ps, _ := server.Context().(*PeerServer)
+	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 {
+		if context.CommitIndex() > ps.joinIndex && ps.joinIndex != 0 {
-			log.Debugf("server [%s] is removed", server.Name())
+			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
 }

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) {
+func (c *CompareAndSwapCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	s, _ := context.Server().StateMachine().(store.Store)
 
 	e, err := s.CompareAndSwap(c.Key, c.PrevValue, c.PrevIndex, c.Value, c.ExpireTime)
 

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) {
+func (c *CreateCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	s, _ := context.Server().StateMachine().(store.Store)
 
 	e, err := s.Create(c.Key, c.Dir, c.Value, c.Unique, c.ExpireTime)
 

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) {
+func (c *DeleteCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	s, _ := context.Server().StateMachine().(store.Store)
 
 	if c.Recursive {
 		// recursive implies dir

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) {
+func (c *SetCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	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)

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) {
+func (c SyncCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	s, _ := context.Server().StateMachine().(store.Store)
 	s.DeleteExpiredKeys(c.Time)
 
 	return nil, nil

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) {
+func (c *UpdateCommand) Apply(context raft.Context) (interface{}, error) {
-	s, _ := server.StateMachine().(store.Store)
+	s, _ := context.Server().StateMachine().(store.Store)
 
 	e, err := s.Update(c.Key, c.Value, c.ExpireTime)
 

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())
 	}
 }