vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

refactor return http status 201 Created when creating a new node

release-0.4
Xiang Li 2013-10-19 19:57:36 -07:00
parent 97987e72a7
commit acbc0c8846
4 changed files with 95 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
server/peer_server.go
View File

@ -9,7 +9,6 @@ import (
"io/ioutil"
"net/http"
"net/url"
"strings"
"time"
etcdErr "github.com/coreos/etcd/error"
@ -344,53 +343,3 @@ func (s *PeerServer) monitorSnapshot() {
}
}
}
func (s *PeerServer) dispatch(c raft.Command, w http.ResponseWriter, req *http.Request) error {
if s.raftServer.State() == raft.Leader {
result, err := s.raftServer.Do(c)
if err != nil {
return err
}
if result == nil {
return etcdErr.NewError(300, "Empty result from raft", store.UndefIndex, store.UndefTerm)
}
// response for raft related commands[join/remove]
if b, ok := result.([]byte); ok {
w.WriteHeader(http.StatusOK)
w.Write(b)
return nil
}
var b []byte
if strings.HasPrefix(req.URL.Path, "/v1") {
b, _ = json.Marshal(result.(*store.Event).Response())
} else {
b, _ = json.Marshal(result.(*store.Event))
}
w.WriteHeader(http.StatusOK)
w.Write(b)
return nil
} else {
leader := s.raftServer.Leader()
// No leader available.
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
}
var url string
switch c.(type) {
case *JoinCommand, *RemoveCommand:
url, _ = s.registry.PeerURL(leader)
default:
url, _ = s.registry.ClientURL(leader)
}
redirect(url, w, req)
return nil
}
}

60
server/peer_server_handlers.go
View File

@ -10,20 +10,20 @@ import (
)
// Get all the current logs
func (s *PeerServer) GetLogHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] GET %s/log", s.url)
func (ps *PeerServer) GetLogHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] GET %s/log", ps.url)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(s.raftServer.LogEntries())
json.NewEncoder(w).Encode(ps.raftServer.LogEntries())
}
// Response to vote request
func (s *PeerServer) VoteHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) VoteHttpHandler(w http.ResponseWriter, req *http.Request) {
rvreq := &raft.RequestVoteRequest{}
err := decodeJsonRequest(req, rvreq)
if err == nil {
log.Debugf("[recv] POST %s/vote [%s]", s.url, rvreq.CandidateName)
if resp := s.raftServer.RequestVote(rvreq); resp != nil {
log.Debugf("[recv] POST %s/vote [%s]", ps.url, rvreq.CandidateName)
if resp := ps.raftServer.RequestVote(rvreq); resp != nil {
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(resp)
return
@ -34,16 +34,16 @@ func (s *PeerServer) VoteHttpHandler(w http.ResponseWriter, req *http.Request) {
}
// Response to append entries request
func (s *PeerServer) AppendEntriesHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) AppendEntriesHttpHandler(w http.ResponseWriter, req *http.Request) {
aereq := &raft.AppendEntriesRequest{}
err := decodeJsonRequest(req, aereq)
if err == nil {
log.Debugf("[recv] POST %s/log/append [%d]", s.url, len(aereq.Entries))
log.Debugf("[recv] POST %s/log/append [%d]", ps.url, len(aereq.Entries))
s.serverStats.RecvAppendReq(aereq.LeaderName, int(req.ContentLength))
ps.serverStats.RecvAppendReq(aereq.LeaderName, int(req.ContentLength))
if resp := s.raftServer.AppendEntries(aereq); resp != nil {
if resp := ps.raftServer.AppendEntries(aereq); resp != nil {
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(resp)
if !resp.Success {
@ -57,12 +57,12 @@ func (s *PeerServer) AppendEntriesHttpHandler(w http.ResponseWriter, req *http.R
}
// Response to recover from snapshot request
func (s *PeerServer) SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) {
aereq := &raft.SnapshotRequest{}
err := decodeJsonRequest(req, aereq)
if err == nil {
log.Debugf("[recv] POST %s/snapshot/ ", s.url)
if resp := s.raftServer.RequestSnapshot(aereq); resp != nil {
log.Debugf("[recv] POST %s/snapshot/ ", ps.url)
if resp := ps.raftServer.RequestSnapshot(aereq); resp != nil {
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(resp)
return
@ -73,12 +73,12 @@ func (s *PeerServer) SnapshotHttpHandler(w http.ResponseWriter, req *http.Reques
}
// Response to recover from snapshot request
func (s *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *http.Request) {
aereq := &raft.SnapshotRecoveryRequest{}
err := decodeJsonRequest(req, aereq)
if err == nil {
log.Debugf("[recv] POST %s/snapshotRecovery/ ", s.url)
if resp := s.raftServer.SnapshotRecoveryRequest(aereq); resp != nil {
log.Debugf("[recv] POST %s/snapshotRecovery/ ", ps.url)
if resp := ps.raftServer.SnapshotRecoveryRequest(aereq); resp != nil {
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(resp)
return
@ -89,20 +89,20 @@ func (s *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *htt
}
// Get the port that listening for etcd connecting of the server
func (s *PeerServer) EtcdURLHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/etcdURL/ ", s.url)
func (ps *PeerServer) EtcdURLHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/etcdURL/ ", ps.url)
w.WriteHeader(http.StatusOK)
w.Write([]byte(s.server.URL()))
w.Write([]byte(ps.server.URL()))
}
// Response to the join request
func (s *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
command := &JoinCommand{}
// Write CORS header.
if s.server.OriginAllowed("*") {
if ps.server.OriginAllowed("*") {
w.Header().Add("Access-Control-Allow-Origin", "*")
} else if s.server.OriginAllowed(req.Header.Get("Origin")) {
} else if ps.server.OriginAllowed(req.Header.Get("Origin")) {
w.Header().Add("Access-Control-Allow-Origin", req.Header.Get("Origin"))
}
@ -113,7 +113,7 @@ func (s *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
}
log.Debugf("Receive Join Request from %s", command.Name)
err = s.dispatch(command, w, req)
err = ps.server.Dispatch(command, w, req)
// Return status.
if err != nil {
@ -127,7 +127,7 @@ func (s *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
}
// Response to remove request
func (s *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request) {
func (ps *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request) {
if req.Method != "DELETE" {
w.WriteHeader(http.StatusMethodNotAllowed)
return
@ -140,19 +140,19 @@ func (s *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request)
log.Debugf("[recv] Remove Request [%s]", command.Name)
s.dispatch(command, w, req)
ps.server.Dispatch(command, w, req)
}
// Response to the name request
func (s *PeerServer) NameHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/name/ ", s.url)
func (ps *PeerServer) NameHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/name/ ", ps.url)
w.WriteHeader(http.StatusOK)
w.Write([]byte(s.name))
w.Write([]byte(ps.name))
}
// Response to the name request
func (s *PeerServer) RaftVersionHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/version/ ", s.url)
func (ps *PeerServer) RaftVersionHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/version/ ", ps.url)
w.WriteHeader(http.StatusOK)
w.Write([]byte(PeerVersion))
}

53
server/server.go
View File

@ -1,6 +1,7 @@
package server
import (
"encoding/json"
"fmt"
"net/http"
"net/url"
@ -167,7 +168,57 @@ func (s *Server) ListenAndServe() {
}
func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Request) error {
return s.peerServer.dispatch(c, w, req)
ps := s.peerServer
if ps.raftServer.State() == raft.Leader {
result, err := ps.raftServer.Do(c)
if err != nil {
return err
}
if result == nil {
return etcdErr.NewError(300, "Empty result from raft", store.UndefIndex, store.UndefTerm)
}
// response for raft related commands[join/remove]
if b, ok := result.([]byte); ok {
w.WriteHeader(http.StatusOK)
w.Write(b)
return nil
}
var b []byte
if strings.HasPrefix(req.URL.Path, "/v1") {
b, _ = json.Marshal(result.(*store.Event).Response())
w.WriteHeader(http.StatusOK)
} else {
e, _ := result.(*store.Event)
b, _ = json.Marshal(e)
w.WriteHeader(e.HttpStatusCode())
}
w.Write(b)
return nil
} else {
leader := ps.raftServer.Leader()
// No leader available.
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
}
var url string
switch c.(type) {
case *JoinCommand, *RemoveCommand:
url, _ = ps.registry.PeerURL(leader)
default:
url, _ = ps.registry.ClientURL(leader)
}
redirect(url, w, req)
return nil
}
}
// Sets a comma-delimited list of origins that are allowed.

13
store/event.go
View File

@ -1,6 +1,7 @@
package store
import (
"net/http"
"time"
)
@ -42,6 +43,18 @@ func newEvent(action string, key string, index uint64, term uint64) *Event {
}
}
func (e *Event) HttpStatusCode() int {
if e.Action == Create {
return http.StatusCreated
}
if e.Action == Set && e.PrevValue == "" {
return http.StatusCreated
}
return http.StatusOK
}
// Converts an event object into a response object.
func (event *Event) Response() interface{} {
if !event.Dir {