package server import ( "encoding/json" "net/http" "net/url" "strconv" "time" "github.com/coreos/etcd/third_party/github.com/goraft/raft" "github.com/coreos/etcd/third_party/github.com/gorilla/mux" etcdErr "github.com/coreos/etcd/error" "github.com/coreos/etcd/log" uhttp "github.com/coreos/etcd/pkg/http" "github.com/coreos/etcd/store" ) // Get all the current logs func (ps *PeerServer) GetLogHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] GET %s/log", ps.Config.URL) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(ps.raftServer.LogEntries()) } // Response to vote request func (ps *PeerServer) VoteHttpHandler(w http.ResponseWriter, req *http.Request) { rvreq := &raft.RequestVoteRequest{} if _, err := rvreq.Decode(req.Body); err != nil { http.Error(w, "", http.StatusBadRequest) log.Warnf("[recv] BADREQUEST %s/vote [%v]", ps.Config.URL, err) return } log.Debugf("[recv] POST %s/vote [%s]", ps.Config.URL, rvreq.CandidateName) resp := ps.raftServer.RequestVote(rvreq) if resp == nil { log.Warn("[vote] Error: nil response") http.Error(w, "", http.StatusInternalServerError) return } if _, err := resp.Encode(w); err != nil { log.Warn("[vote] Error: %v", err) http.Error(w, "", http.StatusInternalServerError) return } } // Response to append entries request func (ps *PeerServer) AppendEntriesHttpHandler(w http.ResponseWriter, req *http.Request) { start := time.Now() aereq := &raft.AppendEntriesRequest{} if _, err := aereq.Decode(req.Body); err != nil { http.Error(w, "", http.StatusBadRequest) log.Warnf("[recv] BADREQUEST %s/log/append [%v]", ps.Config.URL, err) return } log.Debugf("[recv] POST %s/log/append [%d]", ps.Config.URL, len(aereq.Entries)) ps.serverStats.RecvAppendReq(aereq.LeaderName, int(req.ContentLength)) resp := ps.raftServer.AppendEntries(aereq) if resp == nil { log.Warn("[ae] Error: nil response") http.Error(w, "", http.StatusInternalServerError) return } if !resp.Success() { log.Debugf("[Append Entry] Step back") } if _, err := resp.Encode(w); err != nil { log.Warn("[ae] Error: %v", err) http.Error(w, "", http.StatusInternalServerError) return } (*ps.metrics).Timer("timer.appendentries.handle").UpdateSince(start) } // Response to recover from snapshot request func (ps *PeerServer) SnapshotHttpHandler(w http.ResponseWriter, req *http.Request) { ssreq := &raft.SnapshotRequest{} if _, err := ssreq.Decode(req.Body); err != nil { http.Error(w, "", http.StatusBadRequest) log.Warnf("[recv] BADREQUEST %s/snapshot [%v]", ps.Config.URL, err) return } log.Debugf("[recv] POST %s/snapshot", ps.Config.URL) resp := ps.raftServer.RequestSnapshot(ssreq) if resp == nil { log.Warn("[ss] Error: nil response") http.Error(w, "", http.StatusInternalServerError) return } if _, err := resp.Encode(w); err != nil { log.Warn("[ss] Error: %v", err) http.Error(w, "", http.StatusInternalServerError) return } } // Response to recover from snapshot request func (ps *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *http.Request) { ssrreq := &raft.SnapshotRecoveryRequest{} if _, err := ssrreq.Decode(req.Body); err != nil { http.Error(w, "", http.StatusBadRequest) log.Warnf("[recv] BADREQUEST %s/snapshotRecovery [%v]", ps.Config.URL, err) return } log.Debugf("[recv] POST %s/snapshotRecovery", ps.Config.URL) resp := ps.raftServer.SnapshotRecoveryRequest(ssrreq) if resp == nil { log.Warn("[ssr] Error: nil response") http.Error(w, "", http.StatusInternalServerError) return } if _, err := resp.Encode(w); err != nil { log.Warn("[ssr] Error: %v", err) http.Error(w, "", http.StatusInternalServerError) return } } // Get the port that listening for etcd connecting of the server func (ps *PeerServer) EtcdURLHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] Get %s/etcdURL/ ", ps.Config.URL) w.WriteHeader(http.StatusOK) w.Write([]byte(ps.server.URL())) } // Response to the join request func (ps *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) { command := &JoinCommandV1{} if err := uhttp.DecodeJsonRequest(req, command); err != nil { w.WriteHeader(http.StatusInternalServerError) return } log.Debugf("Receive Join Request from %s", command.Name) err := ps.server.Dispatch(command, w, req) // Return status. if err != nil { if etcdErr, ok := err.(*etcdErr.Error); ok { log.Debug("Return error: ", (*etcdErr).Error()) etcdErr.Write(w) } else { http.Error(w, err.Error(), http.StatusInternalServerError) } } } // Attempt to rejoin the cluster as a peer. func (ps *PeerServer) PromoteHttpHandler(w http.ResponseWriter, req *http.Request) { log.Infof("%s attempting to promote in cluster: %s", ps.Config.Name, ps.proxyPeerURL) url, err := url.Parse(ps.proxyPeerURL) if err != nil { w.WriteHeader(http.StatusInternalServerError) return } err = ps.joinByPeer(ps.raftServer, url.Host, ps.Config.Scheme) if err != nil { log.Infof("%s error while promoting: %v", ps.Config.Name, err) w.WriteHeader(http.StatusInternalServerError) return } log.Infof("%s promoted in the cluster", ps.Config.Name) w.WriteHeader(http.StatusOK) } // Response to remove request func (ps *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request) { if req.Method != "DELETE" { w.WriteHeader(http.StatusMethodNotAllowed) return } vars := mux.Vars(req) command := &RemoveCommandV1{ Name: vars["name"], } log.Debugf("[recv] Remove Request [%s]", command.Name) ps.server.Dispatch(command, w, req) } // Returns a JSON-encoded cluster configuration. func (ps *PeerServer) getClusterConfigHttpHandler(w http.ResponseWriter, req *http.Request) { json.NewEncoder(w).Encode(&ps.clusterConfig) } // Updates the cluster configuration. func (ps *PeerServer) setClusterConfigHttpHandler(w http.ResponseWriter, req *http.Request) { // Decode map. m := make(map[string]interface{}) if err := json.NewDecoder(req.Body).Decode(&m); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } // Copy config and update fields passed in. config := &ClusterConfig{ ActiveSize: ps.clusterConfig.ActiveSize, PromoteDelay: ps.clusterConfig.PromoteDelay, } if activeSize, ok := m["activeSize"].(float64); ok { config.ActiveSize = int(activeSize) } if promoteDelay, ok := m["promoteDelay"].(float64); ok { config.PromoteDelay = int(promoteDelay) } // Issue command to update. c := &SetClusterConfigCommand{Config: config} log.Debugf("[recv] Update Cluster Config Request") ps.server.Dispatch(c, w, req) json.NewEncoder(w).Encode(&ps.clusterConfig) } // Retrieves a list of peers and proxies. func (ps *PeerServer) getMachinesHttpHandler(w http.ResponseWriter, req *http.Request) { machines := make([]*machineMessage, 0) for _, name := range ps.registry.Peers() { machines = append(machines, ps.getMachineMessage(name)) } for _, name := range ps.registry.Proxies() { machines = append(machines, ps.getMachineMessage(name)) } json.NewEncoder(w).Encode(&machines) } // Retrieve single peer or proxy. func (ps *PeerServer) getMachineHttpHandler(w http.ResponseWriter, req *http.Request) { vars := mux.Vars(req) json.NewEncoder(w).Encode(ps.getMachineMessage(vars["name"])) } func (ps *PeerServer) getMachineMessage(name string) *machineMessage { if ps.registry.PeerExists(name) { clientURL, _ := ps.registry.ClientURL(name) peerURL, _ := ps.registry.PeerURL(name) return &machineMessage{ Name: name, Mode: PeerMode, ClientURL: clientURL, PeerURL: peerURL, } } if ps.registry.ProxyExists(name) { clientURL, _ := ps.registry.ProxyClientURL(name) peerURL, _ := ps.registry.ProxyPeerURL(name) return &machineMessage{ Name: name, Mode: ProxyMode, ClientURL: clientURL, PeerURL: peerURL, } } return nil } // Adds a machine to the cluster. func (ps *PeerServer) addMachineHttpHandler(w http.ResponseWriter, req *http.Request) { c := &JoinCommandV2{} if err := uhttp.DecodeJsonRequest(req, c); err != nil { w.WriteHeader(http.StatusInternalServerError) return } log.Debugf("Receive Join Request (v2) from %s", c.Name) if err := ps.server.Dispatch(c, w, req); err != nil { if etcdErr, ok := err.(*etcdErr.Error); ok { log.Debug("Return error: ", (*etcdErr).Error()) etcdErr.Write(w) } else { http.Error(w, err.Error(), http.StatusInternalServerError) } } } // Removes a machine from the cluster. func (ps *PeerServer) removeMachineHttpHandler(w http.ResponseWriter, req *http.Request) { vars := mux.Vars(req) c := &RemoveCommandV2{Name: vars["name"]} log.Debugf("[recv] Remove Request [%s]", c.Name) ps.server.Dispatch(c, w, req) } // Response to the name request func (ps *PeerServer) NameHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] Get %s/name/ ", ps.Config.URL) w.WriteHeader(http.StatusOK) w.Write([]byte(ps.Config.Name)) } // Response to the name request func (ps *PeerServer) VersionHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] Get %s/version/ ", ps.Config.URL) w.WriteHeader(http.StatusOK) w.Write([]byte(strconv.Itoa(ps.store.Version()))) } // Checks whether a given version is supported. func (ps *PeerServer) VersionCheckHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] Get %s%s ", ps.Config.URL, req.URL.Path) vars := mux.Vars(req) version, _ := strconv.Atoi(vars["version"]) if version >= store.MinVersion() && version <= store.MaxVersion() { w.WriteHeader(http.StatusOK) } else { w.WriteHeader(http.StatusForbidden) } } // Upgrades the current store version to the next version. func (ps *PeerServer) UpgradeHttpHandler(w http.ResponseWriter, req *http.Request) { log.Debugf("[recv] Get %s/version", ps.Config.URL) // Check if upgrade is possible for all nodes. if err := ps.Upgradable(); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } // Create an upgrade command from the current version. c := ps.store.CommandFactory().CreateUpgradeCommand() if err := ps.server.Dispatch(c, w, req); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.WriteHeader(http.StatusOK) } // machineMessage represents information about a peer or proxy in the registry. type machineMessage struct { Name string `json:"name"` Mode Mode `json:"mode"` ClientURL string `json:"clientURL"` PeerURL string `json:"peerURL"` }