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

Refactor v1 API into server/v1.

release-0.4
Ben Johnson 2013-10-10 22:42:45 -06:00
parent a8b677f6e7
commit a113a51d73
12 changed files with 331 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
etcd.go
View File

@ -45,7 +45,6 @@ var (
cpuprofile string
cors string
corsList map[string]bool
)
func init() {
@ -212,20 +211,3 @@ func main() {
}
// parseCorsFlag gathers up the cors whitelist and puts it into the corsList.
func parseCorsFlag() {
if cors != "" {
corsList = make(map[string]bool)
list := strings.Split(cors, ",")
for _, v := range list {
fmt.Println(v)
if v != "*" {
_, err := url.Parse(v)
if err != nil {
panic(fmt.Sprintf("bad cors url: %s", err))
}
}
corsList[v] = true
}
}
}

250
etcd_handler_v1.go
View File

@ -1,250 +0,0 @@
package main
import (
"encoding/json"
"net/http"
"strconv"
etcdErr "github.com/coreos/etcd/error"
"github.com/coreos/etcd/store"
"github.com/coreos/go-raft"
)
//-------------------------------------------------------------------
// Handlers to handle etcd-store related request via etcd url
//-------------------------------------------------------------------
// Multiplex GET/POST/DELETE request to corresponding handlers
func (e *etcdServer) MultiplexerV1(w http.ResponseWriter, req *http.Request) error {
switch req.Method {
case "GET":
return e.GetHttpHandlerV1(w, req)
case "POST":
return e.SetHttpHandlerV1(w, req)
case "PUT":
return e.SetHttpHandlerV1(w, req)
case "DELETE":
return e.DeleteHttpHandlerV1(w, req)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
return nil
}
}
//--------------------------------------
// State sensitive handlers
// Set/Delete will dispatch to leader
//--------------------------------------
// Set Command Handler
func (e *etcdServer) SetHttpHandlerV1(w http.ResponseWriter, req *http.Request) error {
key := req.URL.Path[len("/v1/keys/"):]
debugf("[recv] POST %v/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
req.ParseForm()
value := req.Form.Get("value")
if len(value) == 0 {
return etcdErr.NewError(200, "Set", store.UndefIndex, store.UndefTerm)
}
strDuration := req.Form.Get("ttl")
expireTime, err := durationToExpireTime(strDuration)
if err != nil {
return etcdErr.NewError(202, "Set", store.UndefIndex, store.UndefTerm)
}
if prevValueArr, ok := req.Form["prevValue"]; ok && len(prevValueArr) > 0 {
command := &TestAndSetCommand{
Key: key,
Value: value,
PrevValue: prevValueArr[0],
ExpireTime: expireTime,
}
return dispatchEtcdCommandV1(command, w, req)
} else {
command := &CreateCommand{
Key: key,
Value: value,
ExpireTime: expireTime,
Force: true,
}
return dispatchEtcdCommandV1(command, w, req)
}
}
// Delete Handler
func (e *etcdServer) DeleteHttpHandlerV1(w http.ResponseWriter, req *http.Request) error {
key := req.URL.Path[len("/v1/keys/"):]
debugf("[recv] DELETE %v/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
command := &DeleteCommand{
Key: key,
}
return dispatchEtcdCommandV1(command, w, req)
}
//--------------------------------------
// State non-sensitive handlers
// will not dispatch to leader
// TODO: add sensitive version for these
// command?
//--------------------------------------
// Get Handler
func (e *etcdServer) GetHttpHandlerV1(w http.ResponseWriter, req *http.Request) error {
key := req.URL.Path[len("/v1/keys/"):]
r := e.raftServer
debugf("[recv] GET %s/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
command := &GetCommand{
Key: key,
}
if event, err := command.Apply(r.Server); err != nil {
return err
} else {
event, _ := event.(*store.Event)
response := eventToResponse(event)
bytes, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(bytes)
return nil
}
}
// Watch handler
func (e *etcdServer) WatchHttpHandlerV1(w http.ResponseWriter, req *http.Request) error {
key := req.URL.Path[len("/v1/watch/"):]
command := &WatchCommand{
Key: key,
}
r := e.raftServer
if req.Method == "GET" {
debugf("[recv] GET %s/watch/%s [%s]", e.url, key, req.RemoteAddr)
command.SinceIndex = 0
} else if req.Method == "POST" {
// watch from a specific index
debugf("[recv] POST %s/watch/%s [%s]", e.url, key, req.RemoteAddr)
content := req.FormValue("index")
sinceIndex, err := strconv.ParseUint(string(content), 10, 64)
if err != nil {
return etcdErr.NewError(203, "Watch From Index", store.UndefIndex, store.UndefTerm)
}
command.SinceIndex = sinceIndex
} else {
w.WriteHeader(http.StatusMethodNotAllowed)
return nil
}
if event, err := command.Apply(r.Server); err != nil {
return etcdErr.NewError(500, key, store.UndefIndex, store.UndefTerm)
} else {
event, _ := event.(*store.Event)
response := eventToResponse(event)
bytes, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(bytes)
return nil
}
}
// Dispatch the command to leader
func dispatchEtcdCommandV1(c Command, w http.ResponseWriter, req *http.Request) error {
return dispatchV1(c, w, req, nameToEtcdURL)
}
func dispatchV1(c Command, w http.ResponseWriter, req *http.Request, toURL func(name string) (string, bool)) error {
r := e.raftServer
if r.State() == raft.Leader {
if event, err := r.Do(c); err != nil {
return err
} else {
if event == nil {
return etcdErr.NewError(300, "Empty result from raft", store.UndefIndex, store.UndefTerm)
}
event, _ := event.(*store.Event)
response := eventToResponse(event)
bytes, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(bytes)
return nil
}
} else {
leader := r.Leader()
// current no leader
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
}
url, _ := toURL(leader)
redirect(url, w, req)
return nil
}
}
func eventToResponse(event *store.Event) interface{} {
if !event.Dir {
response := &store.Response{
Action: event.Action,
Key: event.Key,
Value: event.Value,
PrevValue: event.PrevValue,
Index: event.Index,
TTL: event.TTL,
Expiration: event.Expiration,
}
if response.Action == store.Create || response.Action == store.Update {
response.Action = "set"
if response.PrevValue == "" {
response.NewKey = true
}
}
return response
} else {
responses := make([]*store.Response, len(event.KVPairs))
for i, kv := range event.KVPairs {
responses[i] = &store.Response{
Action: event.Action,
Key: kv.Key,
Value: kv.Value,
Dir: kv.Dir,
Index: event.Index,
}
}
return responses
}
}

10
etcd_handlers.go
View File

@ -18,11 +18,11 @@ import (
func NewEtcdMuxer() *http.ServeMux {
// external commands
etcdMux := http.NewServeMux()
etcdMux.Handle("/"+version+"/keys/", errorHandler(e.Multiplexer))
etcdMux.Handle("/"+version+"/leader", errorHandler(e.LeaderHttpHandler))
etcdMux.Handle("/"+version+"/machines", errorHandler(e.MachinesHttpHandler))
etcdMux.Handle("/"+version+"/stats/", errorHandler(e.StatsHttpHandler))
router := mux.NewRouter()
etcdMux.Handle("/v2/keys/", errorHandler(e.Multiplexer))
etcdMux.Handle("/v2/leader", errorHandler(e.LeaderHttpHandler))
etcdMux.Handle("/v2/machines", errorHandler(e.MachinesHttpHandler))
etcdMux.Handle("/v2/stats/", errorHandler(e.StatsHttpHandler))
etcdMux.Handle("/version", errorHandler(e.VersionHttpHandler))
etcdMux.HandleFunc("/test/", TestHttpHandler)

44
etcd_server.go
View File

@ -1,44 +0,0 @@
package main
import (
"net/http"
)
type etcdServer struct {
http.Server
raftServer *raftServer
name string
url string
tlsConf *TLSConfig
tlsInfo *TLSInfo
}
var e *etcdServer
func newEtcdServer(name string, urlStr string, listenHost string, tlsConf *TLSConfig, tlsInfo *TLSInfo, raftServer *raftServer) *etcdServer {
e = &etcdServer{
Server: http.Server{
TLSConfig: &tlsConf.Server,
Addr: listenHost,
},
name: name,
url: urlStr,
tlsConf: tlsConf,
tlsInfo: tlsInfo,
raftServer: raftServer,
}
e.Handler = NewEtcdMuxer()
return e
}
// Start to listen and response etcd client command
func (e *etcdServer) ListenAndServe() {
infof("etcd server [name %s, listen on %s, advertised url %s]", e.name, e.Server.Addr, e.url)
if e.tlsConf.Scheme == "http" {
fatal(e.Server.ListenAndServe())
} else {
fatal(e.Server.ListenAndServeTLS(e.tlsInfo.CertFile, e.tlsInfo.KeyFile))
}
}

98
server/server.go Normal file
View File

@ -0,0 +1,98 @@
package server
import (
"github.com/gorilla/mux"
"net/http"
)
// The Server provides an HTTP interface to the underlying data store.
type Server struct {
http.Server
raftServer *raftServer
name string
url string
tlsConf *TLSConfig
tlsInfo *TLSInfo
corsOrigins map[string]bool
}
// Creates a new Server.
func New(name string, urlStr string, listenHost string, tlsConf *TLSConfig, tlsInfo *TLSInfo, raftServer *raftServer) *Server {
s := &etcdServer{
Server: http.Server{
Handler: mux.NewRouter(),
TLSConfig: &tlsConf.Server,
Addr: listenHost,
},
name: name,
url: urlStr,
tlsConf: tlsConf,
tlsInfo: tlsInfo,
raftServer: raftServer,
}
// TODO: Move to main.go.
// Install the routes for each version of the API.
// v1.Install(s)
// v2.Install(s)
return s
}
// Adds a server handler to the router.
func (s *Server) HandleFunc(path string, f func(http.ResponseWriter, *http.Request, *server.Server) error) *mux.Route {
r := s.Handler.(*mux.Router)
// Wrap the standard HandleFunc interface to pass in the server reference.
return r.HandleFunc(path, func(w http.ResponseWriter, req *http.Request) {
// Write CORS header.
if s.OriginAllowed("*") {
w.Header().Add("Access-Control-Allow-Origin", "*")
} else if s.OriginAllowed(r.Header.Get("Origin")) {
w.Header().Add("Access-Control-Allow-Origin", r.Header.Get("Origin"))
}
// Execute handler function and return error if necessary.
if err := f(w, req, s); err != nil {
if etcdErr, ok := err.(*etcdErr.Error); ok {
debug("Return error: ", (*etcdErr).Error())
etcdErr.Write(w)
} else {
http.Error(w, e.Error(), http.StatusInternalServerError)
}
}
})
}
// Start to listen and response etcd client command
func (s *Server) ListenAndServe() {
infof("etcd server [name %s, listen on %s, advertised url %s]", s.name, s.Server.Addr, s.url)
if s.tlsConf.Scheme == "http" {
fatal(s.Server.ListenAndServe())
} else {
fatal(s.Server.ListenAndServeTLS(s.tlsInfo.CertFile, s.tlsInfo.KeyFile))
}
}
// Sets a comma-delimited list of origins that are allowed.
func (s *Server) AllowOrigins(origins string) error {
// Construct a lookup of all origins.
m := make(map[string]bool)
for _, v := range strings.Split(cors, ",") {
if v != "*" {
if _, err := url.Parse(v); err != nil {
return fmt.Errorf("Invalid CORS origin: %s", err)
}
}
m[v] = true
}
s.origins = m
return nil
}
// Determines whether the server will allow a given CORS origin.
func (s *Server) OriginAllowed(origin string) {
return s.origins["*"] || s.origins[origin]
}

13
server/v1/delete_key_handler.go Normal file
View File

@ -0,0 +1,13 @@
package v1
func deleteKeyHandler(w http.ResponseWriter, req *http.Request, e *etcdServer) error {
key := req.URL.Path[len("/v1/keys/"):]
debugf("[recv] DELETE %v/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
command := &DeleteCommand{
Key: key,
}
return dispatchEtcdCommandV1(command, w, req)
}

42
server/v1/dispatch.go Normal file
View File

@ -0,0 +1,42 @@
package v1
// Dispatch the command to leader.
func dispatchCommand(c Command, w http.ResponseWriter, req *http.Request) error {
return dispatch(c, w, req, nameToEtcdURL)
}
// Dispatches a command to a given URL.
func dispatch(c Command, w http.ResponseWriter, req *http.Request, toURL func(name string) (string, bool)) error {
r := e.raftServer
if r.State() == raft.Leader {
if event, err := r.Do(c); err != nil {
return err
} else {
if event == nil {
return etcdErr.NewError(300, "Empty result from raft", store.UndefIndex, store.UndefTerm)
}
event, _ := event.(*store.Event)
response := eventToResponse(event)
bytes, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(bytes)
return nil
}
} else {
leader := r.Leader()
// current no leader
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
}
url, _ := toURL(leader)
redirect(url, w, req)
return nil
}
}

31
server/v1/get_key_handler.go Normal file
View File

@ -0,0 +1,31 @@
package v1
import (
"encoding/json"
"github.com/coreos/etcd/store"
"net/http"
)
// Retrieves the value for a given key.
func getKeyHandler(w http.ResponseWriter, req *http.Request, e *etcdServer) error {
vars := mux.Vars(req)
key := "/" + vars["key"]
debugf("[recv] GET %s/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
// Execute the command.
command := &GetCommand{Key: key}
event, err := command.Apply(e.raftServer.Server)
if err != nil {
return err
}
// Convert event to a response and write to client.
event, _ := event.(*store.Event)
response := eventToResponse(event)
b, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(b)
return nil
}

15
server/v1/install.go Normal file
View File

@ -0,0 +1,15 @@
package v1
import (
"github.com/coreos/etcd/server"
"github.com/gorilla/mux"
)
// Installs the routes for version 1 of the API on to a server.
func Install(s *server.Server) {
s.HandleFunc("/v1/keys/{key:.*}", getKeyHandler).Methods("GET")
s.HandleFunc("/v1/keys/{key:.*}", setKeyHandler).Methods("POST", "PUT")
s.HandleFunc("/v1/keys/{key:.*}", deleteKeyHandler).Methods("DELETE")
s.HandleFunc("/v1/watch/{key:.*}", watchKeyHandler).Methods("GET", "POST")
}

50
server/v1/set_key_handler.go Normal file
View File

@ -0,0 +1,50 @@
package v1
import (
"encoding/json"
"github.com/coreos/etcd/store"
"net/http"
)
// Sets the value for a given key.
func setKeyHandler(w http.ResponseWriter, req *http.Request, e *etcdServer) error {
vars := mux.Vars(req)
key := "/" + vars["key"]
debugf("[recv] POST %v/v1/keys/%s [%s]", e.url, key, req.RemoteAddr)
req.ParseForm()
// Parse non-blank value.
value := req.Form.Get("value")
if len(value) == 0 {
return error.NewError(200, "Set", store.UndefIndex, store.UndefTerm)
}
// Convert time-to-live to an expiration time.
expireTime, err := durationToExpireTime(req.Form.Get("ttl"))
if err != nil {
return etcdErr.NewError(202, "Set", store.UndefIndex, store.UndefTerm)
}
// If the "prevValue" is specified then test-and-set. Otherwise create a new key.
var c command.Command
if prevValueArr, ok := req.Form["prevValue"]; ok && len(prevValueArr) > 0 {
c = &TestAndSetCommand{
Key: key,
Value: value,
PrevValue: prevValueArr[0],
ExpireTime: expireTime,
}
} else {
c = &CreateCommand{
Key: key,
Value: value,
ExpireTime: expireTime,
Force: true,
}
}
return dispatchEtcdCommand(command, w, req)
}

38
server/v1/util.go Normal file
View File

@ -0,0 +1,38 @@
package v1
// Converts an event object into a response object.
func eventToResponse(event *store.Event) interface{} {
if !event.Dir {
response := &store.Response{
Action: event.Action,
Key: event.Key,
Value: event.Value,
PrevValue: event.PrevValue,
Index: event.Index,
TTL: event.TTL,
Expiration: event.Expiration,
}
if response.Action == store.Create || response.Action == store.Update {
response.Action = "set"
if response.PrevValue == "" {
response.NewKey = true
}
}
return response
} else {
responses := make([]*store.Response, len(event.KVPairs))
for i, kv := range event.KVPairs {
responses[i] = &store.Response{
Action: event.Action,
Key: kv.Key,
Value: kv.Value,
Dir: kv.Dir,
Index: event.Index,
}
}
return responses
}
}

39
server/v1/watch_key_handler.go Normal file
View File

@ -0,0 +1,39 @@
package v1
import (
"encoding/json"
"github.com/coreos/etcd/store"
"net/http"
)
// Watches a given key prefix for changes.
func watchKeyHandler(w http.ResponseWriter, req *http.Request, e *etcdServer) error {
vars := mux.Vars(req)
key := "/" + vars["key"]
debugf("[recv] %s %s/watch/%s [%s]", req.Method, e.url, key, req.RemoteAddr)
// Create a command to watch from a given index (default 0).
command := &WatchCommand{Key: key}
if req.Method == "POST" {
sinceIndex, err := strconv.ParseUint(string(req.FormValue("index")), 10, 64)
if err != nil {
return etcdErr.NewError(203, "Watch From Index", store.UndefIndex, store.UndefTerm)
}
command.SinceIndex = sinceIndex
}
// Apply the command and write the response.
event, err := command.Apply(e.raftServer.Server)
if err != nil {
return etcdErr.NewError(500, key, store.UndefIndex, store.UndefTerm)
}
event, _ := event.(*store.Event)
response := eventToResponse(event)
b, _ := json.Marshal(response)
w.WriteHeader(http.StatusOK)
w.Write(b)
return nil
}