Merge pull request #27 from xiangli-cmu/snapshot

Add Snapshot handler

client_handlers.go

@@ -110,7 +110,7 @@ func dispatch(c Command, w *http.ResponseWriter, req *http.Request, client bool)
 				return
 			}
 			(*w).WriteHeader(http.StatusInternalServerError)
-			(*w).Write(newJsonError(300, "No Leader"))
+			(*w).Write(newJsonError(300, err.Error()))
 			return
 		} else {
 
@@ -118,6 +118,7 @@ func dispatch(c Command, w *http.ResponseWriter, req *http.Request, client bool)
 				http.NotFound((*w), req)
 			} else {
 				body, ok := body.([]byte)
+				// this should not happen
 				if !ok {
 					panic("wrong type")
 				}

command.go

@@ -120,6 +120,5 @@ func (c *JoinCommand) CommandName() string {
 func (c *JoinCommand) Apply(raftServer *raft.Server) (interface{}, error) {
 	err := raftServer.AddPeer(c.Name)
 	addMachine(c.Name, c.Hostname, c.RaftPort, c.ClientPort)
-
 	return []byte("join success"), err
 }

etcd.go

@@ -52,6 +52,8 @@ var ignore bool
 
 var maxSize int
 
+var snapshot bool
+
 func init() {
 	flag.BoolVar(&verbose, "v", false, "verbose logging")
 
@@ -75,6 +77,8 @@ func init() {
 
 	flag.BoolVar(&ignore, "i", false, "ignore the old configuration, create a new node")
 
+	flag.BoolVar(&snapshot, "snapshot", false, "open or close snapshot")
+
 	flag.IntVar(&maxSize, "m", 1024, "the max size of result buffer")
 }
 
@@ -207,13 +211,15 @@ func startRaft(securityType int) {
 	}
 
 	// LoadSnapshot
-	// err = raftServer.LoadSnapshot()
+	if snapshot {
+		err = raftServer.LoadSnapshot()
 
-	// if err == nil {
-	// 	debug("%s finished load snapshot", raftServer.Name())
-	// } else {
-	// 	debug(err)
-	// }
+		if err == nil {
+			debug("%s finished load snapshot", raftServer.Name())
+		} else {
+			debug(err.Error())
+		}
+	}
 
 	raftServer.Initialize()
 	raftServer.SetElectionTimeout(ELECTIONTIMTOUT)
@@ -249,7 +255,7 @@ func startRaft(securityType int) {
 
 				err = joinCluster(raftServer, machine)
 				if err != nil {
-					debug("cannot join to cluster via machine %s", machine)
+					debug("cannot join to cluster via machine %s %s", machine, err)
 				} else {
 					break
 				}
@@ -267,7 +273,9 @@ func startRaft(securityType int) {
 	}
 
 	// open the snapshot
-	// go server.Snapshot()
+	if snapshot {
+		go raftServer.Snapshot()
+	}
 
 	// start to response to raft requests
 	go startRaftTransport(info.RaftPort, securityType)
@@ -332,6 +340,7 @@ func startRaftTransport(port int, st int) {
 	http.HandleFunc("/log", GetLogHttpHandler)
 	http.HandleFunc("/log/append", AppendEntriesHttpHandler)
 	http.HandleFunc("/snapshot", SnapshotHttpHandler)
+	http.HandleFunc("/snapshotRecovery", SnapshotRecoveryHttpHandler)
 	http.HandleFunc("/client", ClientHttpHandler)
 
 	switch st {

raft_handlers.go

@@ -13,7 +13,7 @@ import (
 
 // Get all the current logs
 func GetLogHttpHandler(w http.ResponseWriter, req *http.Request) {
-	debug("[recv] GET http://%v/log", raftServer.Name())
+	debug("[recv] GET %s/log", raftTransporter.scheme+raftServer.Name())
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(raftServer.LogEntries())
@@ -24,7 +24,7 @@ func VoteHttpHandler(w http.ResponseWriter, req *http.Request) {
 	rvreq := &raft.RequestVoteRequest{}
 	err := decodeJsonRequest(req, rvreq)
 	if err == nil {
-		debug("[recv] POST http://%v/vote [%s]", raftServer.Name(), rvreq.CandidateName)
+		debug("[recv] POST %s/vote [%s]", raftTransporter.scheme+raftServer.Name(), rvreq.CandidateName)
 		if resp := raftServer.RequestVote(rvreq); resp != nil {
 			w.WriteHeader(http.StatusOK)
 			json.NewEncoder(w).Encode(resp)
@@ -41,7 +41,7 @@ func AppendEntriesHttpHandler(w http.ResponseWriter, req *http.Request) {
 	err := decodeJsonRequest(req, aereq)
 
 	if err == nil {
-		debug("[recv] POST http://%s/log/append [%d]", raftServer.Name(), len(aereq.Entries))
+		debug("[recv] POST %s/log/append [%d]", raftTransporter.scheme+raftServer.Name(), len(aereq.Entries))
 		if resp := raftServer.AppendEntries(aereq); resp != nil {
 			w.WriteHeader(http.StatusOK)
 			json.NewEncoder(w).Encode(resp)
@@ -60,8 +60,24 @@ func SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) {
 	aereq := &raft.SnapshotRequest{}
 	err := decodeJsonRequest(req, aereq)
 	if err == nil {
-		debug("[recv] POST http://%s/snapshot/ ", raftServer.Name())
-		if resp, _ := raftServer.SnapshotRecovery(aereq); resp != nil {
+		debug("[recv] POST %s/snapshot/ ", raftTransporter.scheme+raftServer.Name())
+		if resp := raftServer.RequestSnapshot(aereq); resp != nil {
+			w.WriteHeader(http.StatusOK)
+			json.NewEncoder(w).Encode(resp)
+			return
+		}
+	}
+	warn("[Snapshot] ERROR: %v", err)
+	w.WriteHeader(http.StatusInternalServerError)
+}
+
+// Response to recover from snapshot request
+func SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *http.Request) {
+	aereq := &raft.SnapshotRecoveryRequest{}
+	err := decodeJsonRequest(req, aereq)
+	if err == nil {
+		debug("[recv] POST %s/snapshotRecovery/ ", raftTransporter.scheme+raftServer.Name())
+		if resp := raftServer.SnapshotRecoveryRequest(aereq); resp != nil {
 			w.WriteHeader(http.StatusOK)
 			json.NewEncoder(w).Encode(resp)
 			return
@@ -73,7 +89,7 @@ func SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) {
 
 // Get the port that listening for client connecting of the server
 func ClientHttpHandler(w http.ResponseWriter, req *http.Request) {
-	debug("[recv] Get http://%v/client/ ", raftServer.Name())
+	debug("[recv] Get %s/client/ ", raftTransporter.scheme+raftServer.Name())
 	w.WriteHeader(http.StatusOK)
 	client := hostname + ":" + strconv.Itoa(clientPort)
 	w.Write([]byte(client))

transporter.go

@@ -89,6 +89,27 @@ func (t transporter) SendSnapshotRequest(server *raft.Server, peer *raft.Peer, r
 	return aersp
 }
 
+// Sends SnapshotRecoveryRequest RPCs to a peer when the server is the candidate.
+func (t transporter) SendSnapshotRecoveryRequest(server *raft.Server, peer *raft.Peer, req *raft.SnapshotRecoveryRequest) *raft.SnapshotRecoveryResponse {
+	var aersp *raft.SnapshotRecoveryResponse
+	var b bytes.Buffer
+	json.NewEncoder(&b).Encode(req)
+
+	debug("Send SnapshotRecovery to %s [Last Term: %d, LastIndex %d]", peer.Name(),
+		req.LastTerm, req.LastIndex)
+
+	resp, err := t.Post(fmt.Sprintf("%s/snapshotRecovery", peer.Name()), &b)
+
+	if resp != nil {
+		defer resp.Body.Close()
+		aersp = &raft.SnapshotRecoveryResponse{}
+		if err = json.NewDecoder(resp.Body).Decode(&aersp); err == nil || err == io.EOF {
+			return aersp
+		}
+	}
+	return aersp
+}
+
 // Get the client address of the leader in the cluster
 func (t transporter) GetLeaderClientAddress() string {
 	resp, _ := t.Get(raftServer.Leader() + "/client")