Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #292 from xiangli-cmu/fix-ttl

Browse Source 
WIP: fix ttl
This commit is contained in:
Xiang Li
2013-11-11 21:30:32 -08:00
parent fc25ef8383 811c577fe8
commit 5abbaf59e3
40 changed files with 728 additions and 506 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
server/join_command.go
View File

@@ -14,20 +14,20 @@ func init() {
// The JoinCommand adds a node to the cluster.
type JoinCommand struct {
MinVersion int `json:"minVersion"`
MaxVersion int `json:"maxVersion"`
Name string `json:"name"`
RaftURL string `json:"raftURL"`
EtcdURL string `json:"etcdURL"`
MinVersion int `json:"minVersion"`
MaxVersion int `json:"maxVersion"`
Name string `json:"name"`
RaftURL string `json:"raftURL"`
EtcdURL string `json:"etcdURL"`
}
func NewJoinCommand(minVersion int, maxVersion int, name, raftUrl, etcdUrl string) *JoinCommand {
return &JoinCommand{
MinVersion: minVersion,
MaxVersion: maxVersion,
Name: name,
RaftURL: raftUrl,
EtcdURL: etcdUrl,
Name: name,
RaftURL: raftUrl,
EtcdURL: etcdUrl,
}
}
@@ -54,11 +54,11 @@ func (c *JoinCommand) Apply(server raft.Server) (interface{}, error) {
// Check machine number in the cluster
if ps.registry.Count() == ps.MaxClusterSize {
log.Debug("Reject join request from ", c.Name)
return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMoreMachine, "", server.CommitIndex(), server.Term())
return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMoreMachine, "", server.CommitIndex())
}
// Add to shared machine registry.
ps.registry.Register(c.Name, c.RaftURL, c.EtcdURL, server.CommitIndex(), server.Term())
ps.registry.Register(c.Name, c.RaftURL, c.EtcdURL)
// Add peer in raft
err := server.AddPeer(c.Name, "")

14
server/peer_server.go
View File

@@ -136,6 +136,8 @@ func (s *PeerServer) ListenAndServe(snapshot bool, cluster []string) error {
log.Debugf("%s restart as a follower", s.name)
}
go s.monitorSync()
// open the snapshot
if snapshot {
go s.monitorSnapshot()
@@ -424,3 +426,15 @@ func (s *PeerServer) monitorSnapshot() {
}
}
}
func (s *PeerServer) monitorSync() {
ticker := time.Tick(time.Millisecond * 500)
for {
select {
case now := <-ticker:
if s.raftServer.State() == raft.Leader {
s.raftServer.Do(s.store.CommandFactory().CreateSyncCommand(now))
}
}
}
}

18
server/registry.go
View File

@@ -38,20 +38,20 @@ func NewRegistry(s store.Store) *Registry {
}
// Adds a node to the registry.
func (r *Registry) Register(name string, peerURL string, url string, commitIndex uint64, term uint64) error {
func (r *Registry) Register(name string, peerURL string, url string) error {
r.Lock()
defer r.Unlock()
// Write data to store.
key := path.Join(RegistryKey, name)
value := fmt.Sprintf("raft=%s&etcd=%s", peerURL, url)
_, err := r.store.Create(key, value, false, store.Permanent, commitIndex, term)
_, err := r.store.Create(key, value, false, store.Permanent)
log.Debugf("Register: %s", name)
return err
}
// Removes a node from the registry.
func (r *Registry) Unregister(name string, commitIndex uint64, term uint64) error {
func (r *Registry) Unregister(name string) error {
r.Lock()
defer r.Unlock()
@@ -59,14 +59,14 @@ func (r *Registry) Unregister(name string, commitIndex uint64, term uint64) erro
// delete(r.nodes, name)
// Remove the key from the store.
_, err := r.store.Delete(path.Join(RegistryKey, name), false, commitIndex, term)
_, err := r.store.Delete(path.Join(RegistryKey, name), false)
log.Debugf("Unregister: %s", name)
return err
}
// Returns the number of nodes in the cluster.
func (r *Registry) Count() int {
e, err := r.store.Get(RegistryKey, false, false, 0, 0)
e, err := r.store.Get(RegistryKey, false, false)
if err != nil {
return 0
}
@@ -133,7 +133,7 @@ func (r *Registry) urls(leaderName, selfName string, url func(name string) (stri
}
// Retrieve a list of all nodes.
if e, _ := r.store.Get(RegistryKey, false, false, 0, 0); e != nil {
if e, _ := r.store.Get(RegistryKey, false, false); e != nil {
// Lookup the URL for each one.
for _, pair := range e.KVPairs {
_, name := filepath.Split(pair.Key)
@@ -160,7 +160,7 @@ func (r *Registry) load(name string) {
}
// Retrieve from store.
e, err := r.store.Get(path.Join(RegistryKey, name), false, false, 0, 0)
e, err := r.store.Get(path.Join(RegistryKey, name), false, false)
if err != nil {
return
}
@@ -173,7 +173,7 @@ func (r *Registry) load(name string) {
// Create node.
r.nodes[name] = &node{
url: m["etcd"][0],
peerURL: m["raft"][0],
url: m["etcd"][0],
peerURL: m["raft"][0],
}
}

2
server/remove_command.go
View File

@@ -27,7 +27,7 @@ func (c *RemoveCommand) Apply(server raft.Server) (interface{}, error) {
ps, _ := server.Context().(*PeerServer)
// Remove node from the shared registry.
err := ps.registry.Unregister(c.Name, server.CommitIndex(), server.Term())
err := ps.registry.Unregister(c.Name)
// Delete from stats
delete(ps.followersStats.Followers, c.Name)

15
server/server.go
View File

@@ -232,6 +232,7 @@ func (s *Server) Close() {
}
}
// Dispatch command to the current leader
func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Request) error {
ps := s.peerServer
if ps.raftServer.State() == raft.Leader {
@@ -241,7 +242,7 @@ func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Reque
}
if result == nil {
return etcdErr.NewError(300, "Empty result from raft", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(300, "Empty result from raft", s.Store().Index())
}
// response for raft related commands[join/remove]
@@ -259,6 +260,12 @@ func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Reque
e, _ := result.(*store.Event)
b, _ = json.Marshal(e)
// etcd index should be the same as the event index
// which is also the last modified index of the node
w.Header().Add("X-Etcd-Index", fmt.Sprint(e.Index))
w.Header().Add("X-Raft-Index", fmt.Sprint(s.CommitIndex()))
w.Header().Add("X-Raft-Term", fmt.Sprint(s.Term()))
if e.IsCreated() {
w.WriteHeader(http.StatusCreated)
} else {
@@ -275,7 +282,7 @@ func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Reque
// No leader available.
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(300, "", s.Store().Index())
}
var url string
@@ -324,7 +331,7 @@ func (s *Server) GetVersionHandler(w http.ResponseWriter, req *http.Request) err
func (s *Server) GetLeaderHandler(w http.ResponseWriter, req *http.Request) error {
leader := s.peerServer.RaftServer().Leader()
if leader == "" {
return etcdErr.NewError(etcdErr.EcodeLeaderElect, "", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeLeaderElect, "", s.Store().Index())
}
w.WriteHeader(http.StatusOK)
url, _ := s.registry.PeerURL(leader)
@@ -355,7 +362,7 @@ func (s *Server) GetLeaderStatsHandler(w http.ResponseWriter, req *http.Request)
leader := s.peerServer.RaftServer().Leader()
if leader == "" {
return etcdErr.NewError(300, "", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(300, "", s.Store().Index())
}
hostname, _ := s.registry.ClientURL(leader)
redirect(hostname, w, req)

2
server/v1/get_key_handler.go
View File

@@ -13,7 +13,7 @@ func GetKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
key := "/" + vars["key"]
// Retrieve the key from the store.
event, err := s.Store().Get(key, false, false, s.CommitIndex(), s.Term())
event, err := s.Store().Get(key, false, false)
if err != nil {
return err
}

4
server/v1/set_key_handler.go
View File

@@ -19,13 +19,13 @@ func SetKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
// Parse non-blank value.
value := req.Form.Get("value")
if len(value) == 0 {
return etcdErr.NewError(200, "Set", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(200, "Set", s.Store().Index())
}
// Convert time-to-live to an expiration time.
expireTime, err := store.TTL(req.Form.Get("ttl"))
if err != nil {
return etcdErr.NewError(202, "Set", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(202, "Set", s.Store().Index())
}
// If the "prevValue" is specified then test-and-set. Otherwise create a new key.

7
server/v1/watch_key_handler.go
View File

@@ -6,7 +6,6 @@ import (
"strconv"
etcdErr "github.com/coreos/etcd/error"
"github.com/coreos/etcd/store"
"github.com/gorilla/mux"
)
@@ -21,14 +20,14 @@ func WatchKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
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)
return etcdErr.NewError(203, "Watch From Index", s.Store().Index())
}
}
// Start the watcher on the store.
c, err := s.Store().Watch(key, false, sinceIndex, s.CommitIndex(), s.Term())
c, err := s.Store().Watch(key, false, sinceIndex)
if err != nil {
return etcdErr.NewError(500, key, store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(500, key, s.Store().Index())
}
event := <-c

15
server/v2/get_handler.go
View File

@@ -41,14 +41,14 @@ func GetHandler(w http.ResponseWriter, req *http.Request, s Server) error {
if waitIndex != "" {
sinceIndex, err = strconv.ParseUint(string(req.FormValue("waitIndex")), 10, 64)
if err != nil {
return etcdErr.NewError(etcdErr.EcodeIndexNaN, "Watch From Index", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeIndexNaN, "Watch From Index", s.Store().Index())
}
}
// Start the watcher on the store.
eventChan, err := s.Store().Watch(key, recursive, sinceIndex, s.CommitIndex(), s.Term())
eventChan, err := s.Store().Watch(key, recursive, sinceIndex)
if err != nil {
return etcdErr.NewError(500, key, store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(500, key, s.Store().Index())
}
cn, _ := w.(http.CloseNotifier)
@@ -62,17 +62,18 @@ func GetHandler(w http.ResponseWriter, req *http.Request, s Server) error {
} else { //get
// Retrieve the key from the store.
event, err = s.Store().Get(key, recursive, sorted, s.CommitIndex(), s.Term())
event, err = s.Store().Get(key, recursive, sorted)
if err != nil {
return err
}
}
w.Header().Add("X-Etcd-Index", fmt.Sprint(event.Index))
w.Header().Add("X-Etcd-Term", fmt.Sprint(event.Term))
w.Header().Add("X-Etcd-Index", fmt.Sprint(s.Store().Index()))
w.Header().Add("X-Raft-Index", fmt.Sprint(s.CommitIndex()))
w.Header().Add("X-Raft-Term", fmt.Sprint(s.Term()))
w.WriteHeader(http.StatusOK)
b, _ := json.Marshal(event)
w.Write(b)
return nil

2
server/v2/post_handler.go
View File

@@ -15,7 +15,7 @@ func PostHandler(w http.ResponseWriter, req *http.Request, s Server) error {
value := req.FormValue("value")
expireTime, err := store.TTL(req.FormValue("ttl"))
if err != nil {
return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Create", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Create", s.Store().Index())
}
c := s.Store().CommandFactory().CreateCreateCommand(key, value, expireTime, true)

8
server/v2/put_handler.go
View File

@@ -22,7 +22,7 @@ func PutHandler(w http.ResponseWriter, req *http.Request, s Server) error {
value := req.Form.Get("value")
expireTime, err := store.TTL(req.Form.Get("ttl"))
if err != nil {
return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Update", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeTTLNaN, "Update", s.Store().Index())
}
_, valueOk := req.Form["prevValue"]
@@ -59,7 +59,7 @@ func PutHandler(w http.ResponseWriter, req *http.Request, s Server) error {
// bad previous index
if err != nil {
return etcdErr.NewError(etcdErr.EcodeIndexNaN, "CompareAndSwap", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeIndexNaN, "CompareAndSwap", s.Store().Index())
}
} else {
prevIndex = 0
@@ -67,7 +67,7 @@ func PutHandler(w http.ResponseWriter, req *http.Request, s Server) error {
if valueOk {
if prevValue == "" {
return etcdErr.NewError(etcdErr.EcodePrevValueRequired, "CompareAndSwap", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodePrevValueRequired, "CompareAndSwap", s.Store().Index())
}
}
@@ -88,7 +88,7 @@ func CreateHandler(w http.ResponseWriter, req *http.Request, s Server, key, valu
func UpdateHandler(w http.ResponseWriter, req *http.Request, s Server, key, value string, expireTime time.Time) error {
// Update should give at least one option
if value == "" && expireTime.Sub(store.Permanent) == 0 {
return etcdErr.NewError(etcdErr.EcodeValueOrTTLRequired, "Update", store.UndefIndex, store.UndefTerm)
return etcdErr.NewError(etcdErr.EcodeValueOrTTLRequired, "Update", s.Store().Index())
}
c := s.Store().CommandFactory().CreateUpdateCommand(key, value, expireTime)

2
server/v2/tests/delete_handler_test.go
View File

@@ -24,6 +24,6 @@ func TestV2DeleteKey(t *testing.T) {
resp, err = tests.DeleteForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), url.Values{})
body := tests.ReadBody(resp)
assert.Nil(t, err, "")
assert.Equal(t, string(body), `{"action":"delete","key":"/foo/bar","prevValue":"XXX","index":4,"term":0}`, "")
assert.Equal(t, string(body), `{"action":"delete","key":"/foo/bar","prevValue":"XXX","modifiedIndex":2}`, "")
})
}

23
server/v2/tests/get_handler_test.go
View File

@@ -27,8 +27,7 @@ func TestV2GetKey(t *testing.T) {
assert.Equal(t, body["action"], "get", "")
assert.Equal(t, body["key"], "/foo/bar", "")
assert.Equal(t, body["value"], "XXX", "")
assert.Equal(t, body["index"], 3, "")
assert.Equal(t, body["term"], 0, "")
assert.Equal(t, body["modifiedIndex"], 1, "")
})
}
@@ -55,7 +54,7 @@ func TestV2GetKeyRecursively(t *testing.T) {
assert.Equal(t, body["action"], "get", "")
assert.Equal(t, body["key"], "/foo", "")
assert.Equal(t, body["dir"], true, "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["modifiedIndex"], 1, "")
assert.Equal(t, len(body["kvs"].([]interface{})), 2, "")
kv0 := body["kvs"].([]interface{})[0].(map[string]interface{})
@@ -81,9 +80,11 @@ func TestV2GetKeyRecursively(t *testing.T) {
func TestV2WatchKey(t *testing.T) {
tests.RunServer(func(s *server.Server) {
var body map[string]interface{}
c := make(chan bool)
go func() {
resp, _ := tests.Get(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar?wait=true"))
body = tests.ReadBodyJSON(resp)
c <- true
}()
// Make sure response didn't fire early.
@@ -98,12 +99,19 @@ func TestV2WatchKey(t *testing.T) {
// A response should follow from the GET above.
time.Sleep(1 * time.Millisecond)
select {
case <-c:
default:
t.Fatal("cannot get watch result")
}
assert.NotNil(t, body, "")
assert.Equal(t, body["action"], "set", "")
assert.Equal(t, body["key"], "/foo/bar", "")
assert.Equal(t, body["value"], "XXX", "")
assert.Equal(t, body["index"], 3, "")
assert.Equal(t, body["term"], 0, "")
assert.Equal(t, body["modifiedIndex"], 1, "")
})
}
@@ -118,7 +126,7 @@ func TestV2WatchKeyWithIndex(t *testing.T) {
var body map[string]interface{}
c := make(chan bool)
go func() {
resp, _ := tests.Get(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar?wait=true&waitIndex=5"))
resp, _ := tests.Get(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar?wait=true&waitIndex=2"))
body = tests.ReadBodyJSON(resp)
c <- true
}()
@@ -156,7 +164,6 @@ func TestV2WatchKeyWithIndex(t *testing.T) {
assert.Equal(t, body["action"], "set", "")
assert.Equal(t, body["key"], "/foo/bar", "")
assert.Equal(t, body["value"], "YYY", "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["term"], 0, "")
assert.Equal(t, body["modifiedIndex"], 2, "")
})
}

8
server/v2/tests/post_handler_test.go
View File

@@ -21,18 +21,18 @@ func TestV2CreateUnique(t *testing.T) {
resp, _ := tests.PostForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), nil)
body := tests.ReadBodyJSON(resp)
assert.Equal(t, body["action"], "create", "")
assert.Equal(t, body["key"], "/foo/bar/3", "")
assert.Equal(t, body["key"], "/foo/bar/1", "")
assert.Equal(t, body["dir"], true, "")
assert.Equal(t, body["index"], 3, "")
assert.Equal(t, body["modifiedIndex"], 1, "")
// Second POST should add next index to list.
resp, _ = tests.PostForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), nil)
body = tests.ReadBodyJSON(resp)
assert.Equal(t, body["key"], "/foo/bar/4", "")
assert.Equal(t, body["key"], "/foo/bar/2", "")
// POST to a different key should add index to that list.
resp, _ = tests.PostForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/baz"), nil)
body = tests.ReadBodyJSON(resp)
assert.Equal(t, body["key"], "/foo/baz/5", "")
assert.Equal(t, body["key"], "/foo/baz/3", "")
})
}

22
server/v2/tests/put_handler_test.go
View File

@@ -22,7 +22,7 @@ func TestV2SetKey(t *testing.T) {
resp, err := tests.PutForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), v)
body := tests.ReadBody(resp)
assert.Nil(t, err, "")
assert.Equal(t, string(body), `{"action":"set","key":"/foo/bar","value":"XXX","index":3,"term":0}`, "")
assert.Equal(t, string(body), `{"action":"set","key":"/foo/bar","value":"XXX","modifiedIndex":1}`, "")
})
}
@@ -42,7 +42,7 @@ func TestV2SetKeyWithTTL(t *testing.T) {
// Make sure the expiration date is correct.
expiration, _ := time.Parse(time.RFC3339Nano, body["expiration"].(string))
assert.Equal(t, expiration.Sub(t0) / time.Second, 20, "")
assert.Equal(t, expiration.Sub(t0)/time.Second, 20, "")
})
}
@@ -110,7 +110,7 @@ func TestV2UpdateKeySuccess(t *testing.T) {
v.Set("value", "XXX")
resp, _ := tests.PutForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), v)
tests.ReadBody(resp)
v.Set("value", "YYY")
v.Set("prevExist", "true")
resp, _ = tests.PutForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), v)
@@ -160,7 +160,7 @@ func TestV2UpdateKeyFailOnMissingDirectory(t *testing.T) {
// Ensures that a key is set only if the previous index matches.
//
// $ curl -X PUT localhost:4001/v2/keys/foo/bar -d value=XXX
// $ curl -X PUT localhost:4001/v2/keys/foo/bar -d value=YYY -d prevIndex=3
// $ curl -X PUT localhost:4001/v2/keys/foo/bar -d value=YYY -d prevIndex=1
//
func TestV2SetKeyCASOnIndexSuccess(t *testing.T) {
tests.RunServer(func(s *server.Server) {
@@ -169,13 +169,13 @@ func TestV2SetKeyCASOnIndexSuccess(t *testing.T) {
resp, _ := tests.PutForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), v)
tests.ReadBody(resp)
v.Set("value", "YYY")
v.Set("prevIndex", "3")
v.Set("prevIndex", "1")
resp, _ = tests.PutForm(fmt.Sprintf("http://%s%s", s.URL(), "/v2/keys/foo/bar"), v)
body := tests.ReadBodyJSON(resp)
assert.Equal(t, body["action"], "compareAndSwap", "")
assert.Equal(t, body["prevValue"], "XXX", "")
assert.Equal(t, body["value"], "YYY", "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["modifiedIndex"], 2, "")
})
}
@@ -196,8 +196,8 @@ func TestV2SetKeyCASOnIndexFail(t *testing.T) {
body := tests.ReadBodyJSON(resp)
assert.Equal(t, body["errorCode"], 101, "")
assert.Equal(t, body["message"], "Test Failed", "")
assert.Equal(t, body["cause"], "[ != XXX] [10 != 3]", "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["cause"], "[ != XXX] [10 != 1]", "")
assert.Equal(t, body["index"], 1, "")
})
}
@@ -236,7 +236,7 @@ func TestV2SetKeyCASOnValueSuccess(t *testing.T) {
assert.Equal(t, body["action"], "compareAndSwap", "")
assert.Equal(t, body["prevValue"], "XXX", "")
assert.Equal(t, body["value"], "YYY", "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["modifiedIndex"], 2, "")
})
}
@@ -257,8 +257,8 @@ func TestV2SetKeyCASOnValueFail(t *testing.T) {
body := tests.ReadBodyJSON(resp)
assert.Equal(t, body["errorCode"], 101, "")
assert.Equal(t, body["message"], "Test Failed", "")
assert.Equal(t, body["cause"], "[AAA != XXX] [0 != 3]", "")
assert.Equal(t, body["index"], 4, "")
assert.Equal(t, body["cause"], "[AAA != XXX] [0 != 1]", "")
assert.Equal(t, body["index"], 1, "")
})
}