Merge branch 'master' of https://github.com/coreos/etcd into proxy

Conflicts:
	config/config.go
	server/peer_server.go
	server/transporter.go
	tests/server_utils.go

server/listener.go

@@ -3,9 +3,35 @@ package server
 import (
 	"crypto/tls"
 	"net"
+
+	"github.com/coreos/etcd/log"
 )
 
-func NewListener(addr string) (net.Listener, error) {
+// NewListener creates a net.Listener
+// If the given scheme is "https", it will generate TLS configuration based on TLSInfo.
+// If any error happens, this function will call log.Fatal
+func NewListener(scheme, addr string, tlsInfo *TLSInfo) net.Listener {
+	if scheme == "https" {
+		cfg, err := tlsInfo.ServerConfig()
+		if err != nil {
+			log.Fatal("TLS info error: ", err)
+		}
+
+		l, err := newTLSListener(addr, cfg)
+		if err != nil {
+			log.Fatal("Failed to create TLS listener: ", err)
+		}
+		return l
+	}
+
+	l, err := newListener(addr)
+	if err != nil {
+		log.Fatal("Failed to create listener: ", err)
+	}
+	return l
+}
+
+func newListener(addr string) (net.Listener, error) {
 	if addr == "" {
 		addr = ":http"
 	}
@@ -16,7 +42,7 @@ func NewListener(addr string) (net.Listener, error) {
 	return l, nil
 }
 
-func NewTLSListener(addr string, cfg *tls.Config) (net.Listener, error) {
+func newTLSListener(addr string, cfg *tls.Config) (net.Listener, error) {
 	if addr == "" {
 		addr = ":https"
 	}

server/peer_server.go

@@ -231,6 +231,19 @@ func (s *PeerServer) findCluster(discoverURL string, peers []string) {
 	}
 	peers = append(peers, prevPeers...)
 
+	// Remove its own peer address from the peer list to join
+	u, err := url.Parse(s.Config.URL)
+	if err != nil {
+		log.Fatalf("cannot parse peer address %v: %v", s.Config.URL, err)
+	}
+	filteredPeers := make([]string, 0)
+	for _, v := range peers {
+		if v != u.Host {
+			filteredPeers = append(filteredPeers, v)
+		}
+	}
+	peers = filteredPeers
+
 	// if there is backup peer lists, use it to find cluster
 	if len(peers) > 0 {
 		ok := s.joinCluster(peers)
@@ -371,13 +384,12 @@ func (s *PeerServer) startAsFollower(cluster []string) {
 
 // getVersion fetches the peer version of a cluster.
 func getVersion(t *transporter, versionURL url.URL) (int, error) {
-	resp, req, err := t.Get(versionURL.String())
+	resp, _, err := t.Get(versionURL.String())
 	if err != nil {
 		return 0, err
 	}
 	defer resp.Body.Close()
 
-	t.CancelWhenTimeout(req)
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
 		return 0, err
@@ -460,10 +472,11 @@ func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string)
 	json.NewEncoder(&b).Encode(c)
 
 	joinURL := url.URL{Host: peer, Scheme: scheme, Path: "/v2/admin/machines/" + server.Name()}
-	log.Debugf("Send Join Request to %s", joinURL.String())
+	log.Infof("Send Join Request to %s", joinURL.String())
 
 	req, _ := http.NewRequest("PUT", joinURL.String(), &b)
 	resp, err := t.client.Do(req)
+
 	for {
 		if err != nil {
 			return fmt.Errorf("Unable to join: %v", err)
@@ -471,8 +484,7 @@ func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string)
 		if resp != nil {
 			defer resp.Body.Close()
 
-			t.CancelWhenTimeout(req)
-
+			log.Infof("»»»» %d", resp.StatusCode)
 			if resp.StatusCode == http.StatusOK {
 				var msg joinMessageV2
 				if err := json.NewDecoder(resp.Body).Decode(&msg); err != nil {
@@ -500,7 +512,7 @@ func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string)
 					EtcdURL:    s.server.URL(),
 				}
 				json.NewEncoder(&b).Encode(c)
-				resp, req, err = t.Post(address, &b)
+				resp, _, err = t.Post(address, &b)
 
 			} else if resp.StatusCode == http.StatusBadRequest {
 				log.Debug("Reach max number peers in the cluster")

server/transporter.go

@@ -11,6 +11,11 @@ import (
 
 	"github.com/coreos/etcd/log"
 	"github.com/coreos/etcd/third_party/github.com/goraft/raft"
+	httpclient "github.com/coreos/etcd/third_party/github.com/mreiferson/go-httpclient"
+)
+
+const (
+	snapshotTimeout = time.Second * 120
 )
 
 // Transporter layer for communication between raft nodes
@@ -20,8 +25,10 @@ type transporter struct {
 	serverStats    *raftServerStats
 	registry       *Registry
 
-	client    *http.Client
-	transport *http.Transport
+	client            *http.Client
+	transport         *httpclient.Transport
+	snapshotClient    *http.Client
+	snapshotTransport *httpclient.Transport
 }
 
 type dialer func(network, addr string) (net.Conn, error)
@@ -30,20 +37,38 @@ type dialer func(network, addr string) (net.Conn, error)
 // Create http or https transporter based on
 // whether the user give the server cert and key
 func NewTransporter(followersStats *raftFollowersStats, serverStats *raftServerStats, registry *Registry, dialTimeout, requestTimeout, responseHeaderTimeout time.Duration) *transporter {
-	tr := &http.Transport{
-		Dial: func(network, addr string) (net.Conn, error) {
-			return net.DialTimeout(network, addr, dialTimeout)
-		},
+	tr := &httpclient.Transport{
 		ResponseHeaderTimeout: responseHeaderTimeout,
+		// This is a workaround for Transport.CancelRequest doesn't work on
+		// HTTPS connections blocked. The patch for it is in progress,
+		// and would be available in Go1.3
+		// More: https://codereview.appspot.com/69280043/
+		ConnectTimeout:   dialTimeout,
+		RequestTimeout:   dialTimeout + responseHeaderTimeout,
+		ReadWriteTimeout: responseHeaderTimeout,
+	}
+
+	// Sending snapshot might take a long time so we use a different HTTP transporter
+	// Timeout is set to 120s (Around 100MB if the bandwidth is 10Mbits/s)
+	// This timeout is not calculated by heartbeat time.
+	// TODO(xiangl1) we can actually calculate the max bandwidth if we know
+	// average RTT.
+	// It should be equal to (TCP max window size/RTT).
+	sTr := &httpclient.Transport{
+		ConnectTimeout:   dialTimeout,
+		RequestTimeout:   snapshotTimeout,
+		ReadWriteTimeout: snapshotTimeout,
 	}
 
 	t := transporter{
-		client:         &http.Client{Transport: tr},
-		transport:      tr,
-		requestTimeout: requestTimeout,
-		followersStats: followersStats,
-		serverStats:    serverStats,
-		registry:       registry,
+		client:            &http.Client{Transport: tr},
+		transport:         tr,
+		snapshotClient:    &http.Client{Transport: sTr},
+		snapshotTransport: sTr,
+		requestTimeout:    requestTimeout,
+		followersStats:    followersStats,
+		serverStats:       serverStats,
+		registry:          registry,
 	}
 
 	return &t
@@ -52,6 +77,9 @@ func NewTransporter(followersStats *raftFollowersStats, serverStats *raftServerS
 func (t *transporter) SetTLSConfig(tlsConf tls.Config) {
 	t.transport.TLSClientConfig = &tlsConf
 	t.transport.DisableCompression = true
+
+	t.snapshotTransport.TLSClientConfig = &tlsConf
+	t.snapshotTransport.DisableCompression = true
 }
 
 // Sends AppendEntries RPCs to a peer when the server is the leader.
@@ -81,7 +109,7 @@ func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Pe
 
 	start := time.Now()
 
-	resp, httpRequest, err := t.Post(fmt.Sprintf("%s/log/append", u), &b)
+	resp, _, err := t.Post(fmt.Sprintf("%s/log/append", u), &b)
 
 	end := time.Now()
 
@@ -100,8 +128,6 @@ func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Pe
 	if resp != nil {
 		defer resp.Body.Close()
 
-		t.CancelWhenTimeout(httpRequest)
-
 		aeresp := &raft.AppendEntriesResponse{}
 		if _, err = aeresp.Decode(resp.Body); err != nil && err != io.EOF {
 			log.Warn("transporter.ae.decoding.error:", err)
@@ -125,7 +151,7 @@ func (t *transporter) SendVoteRequest(server raft.Server, peer *raft.Peer, req *
 	u, _ := t.registry.PeerURL(peer.Name)
 	log.Debugf("Send Vote from %s to %s", server.Name(), u)
 
-	resp, httpRequest, err := t.Post(fmt.Sprintf("%s/vote", u), &b)
+	resp, _, err := t.Post(fmt.Sprintf("%s/vote", u), &b)
 
 	if err != nil {
 		log.Debugf("Cannot send VoteRequest to %s : %s", u, err)
@@ -134,8 +160,6 @@ func (t *transporter) SendVoteRequest(server raft.Server, peer *raft.Peer, req *
 	if resp != nil {
 		defer resp.Body.Close()
 
-		t.CancelWhenTimeout(httpRequest)
-
 		rvrsp := &raft.RequestVoteResponse{}
 		if _, err = rvrsp.Decode(resp.Body); err != nil && err != io.EOF {
 			log.Warn("transporter.vr.decoding.error:", err)
@@ -158,7 +182,7 @@ func (t *transporter) SendSnapshotRequest(server raft.Server, peer *raft.Peer, r
 	u, _ := t.registry.PeerURL(peer.Name)
 	log.Debugf("Send Snapshot Request from %s to %s", server.Name(), u)
 
-	resp, httpRequest, err := t.Post(fmt.Sprintf("%s/snapshot", u), &b)
+	resp, _, err := t.Post(fmt.Sprintf("%s/snapshot", u), &b)
 
 	if err != nil {
 		log.Debugf("Cannot send Snapshot Request to %s : %s", u, err)
@@ -167,8 +191,6 @@ func (t *transporter) SendSnapshotRequest(server raft.Server, peer *raft.Peer, r
 	if resp != nil {
 		defer resp.Body.Close()
 
-		t.CancelWhenTimeout(httpRequest)
-
 		ssrsp := &raft.SnapshotResponse{}
 		if _, err = ssrsp.Decode(resp.Body); err != nil && err != io.EOF {
 			log.Warn("transporter.ss.decoding.error:", err)
@@ -191,7 +213,7 @@ func (t *transporter) SendSnapshotRecoveryRequest(server raft.Server, peer *raft
 	u, _ := t.registry.PeerURL(peer.Name)
 	log.Debugf("Send Snapshot Recovery from %s to %s", server.Name(), u)
 
-	resp, httpRequest, err := t.Post(fmt.Sprintf("%s/snapshotRecovery", u), &b)
+	resp, err := t.PostSnapshot(fmt.Sprintf("%s/snapshotRecovery", u), &b)
 
 	if err != nil {
 		log.Debugf("Cannot send Snapshot Recovery to %s : %s", u, err)
@@ -200,8 +222,6 @@ func (t *transporter) SendSnapshotRecoveryRequest(server raft.Server, peer *raft
 	if resp != nil {
 		defer resp.Body.Close()
 
-		t.CancelWhenTimeout(httpRequest)
-
 		ssrrsp := &raft.SnapshotRecoveryResponse{}
 		if _, err = ssrrsp.Decode(resp.Body); err != nil && err != io.EOF {
 			log.Warn("transporter.ssr.decoding.error:", err)
@@ -227,10 +247,8 @@ func (t *transporter) Get(urlStr string) (*http.Response, *http.Request, error)
 	return resp, req, err
 }
 
-// Cancel the on fly HTTP transaction when timeout happens.
-func (t *transporter) CancelWhenTimeout(req *http.Request) {
-	go func() {
-		time.Sleep(t.requestTimeout)
-		t.transport.CancelRequest(req)
-	}()
+// PostSnapshot posts a json format snapshot to the given url
+// The underlying HTTP transport has a minute level timeout
+func (t *transporter) PostSnapshot(url string, body io.Reader) (*http.Response, error) {
+	return t.snapshotClient.Post(url, "application/json", body)
 }

server/usage.go

@@ -44,8 +44,8 @@ Peer Communication Options:
   -peer-ca-file=<path>    Path to the peer CA file.
   -peer-cert-file=<path>  Path to the peer cert file.
   -peer-key-file=<path>   Path to the peer key file.
-  -peer-heartbeat-timeout=<time>
-                          Time (in milliseconds) for a heartbeat to timeout.
+  -peer-heartbeat-interval=<time>
+                          Time (in milliseconds) of a heartbeat interval.
   -peer-election-timeout=<time>
                           Time (in milliseconds) for an election to timeout.
 

server/v1/tests/get_handler_test.go

@@ -85,17 +85,15 @@ func TestV1GetKeyDir(t *testing.T) {
 //
 func TestV1WatchKey(t *testing.T) {
 	tests.RunServer(func(s *server.Server) {
-		var body map[string]interface{}
+		var watchResp *http.Response
 		c := make(chan bool)
 		go func() {
-			resp, _ := tests.Get(fmt.Sprintf("%s%s", s.URL(), "/v1/watch/foo/bar"))
-			body = tests.ReadBodyJSON(resp)
+			watchResp, _ = tests.Get(fmt.Sprintf("%s%s", s.URL(), "/v1/watch/foo/bar"))
 			c <- true
 		}()
 
 		// Make sure response didn't fire early.
 		time.Sleep(1 * time.Millisecond)
-		assert.Nil(t, body, "")
 
 		// Set a value.
 		v := url.Values{}
@@ -113,6 +111,7 @@ func TestV1WatchKey(t *testing.T) {
 			t.Fatal("cannot get watch result")
 		}
 
+		body := tests.ReadBodyJSON(watchResp)
 		assert.NotNil(t, body, "")
 		assert.Equal(t, body["action"], "set", "")
 

server/v1/tests/put_handler_test.go

@@ -151,7 +151,7 @@ func TestV1SetKeyCASOnValueFail(t *testing.T) {
 		body := tests.ReadBodyJSON(resp)
 		assert.Equal(t, body["errorCode"], 101, "")
 		assert.Equal(t, body["message"], "Compare failed", "")
-		assert.Equal(t, body["cause"], "[AAA != XXX] [0 != 2]", "")
+		assert.Equal(t, body["cause"], "[AAA != XXX]", "")
 		assert.Equal(t, body["index"], 2, "")
 	})
 }

server/v2/tests/get_handler_test.go

@@ -90,17 +90,15 @@ func TestV2GetKeyRecursively(t *testing.T) {
 //
 func TestV2WatchKey(t *testing.T) {
 	tests.RunServer(func(s *server.Server) {
-		var body map[string]interface{}
+		var watchResp *http.Response
 		c := make(chan bool)
 		go func() {
-			resp, _ := tests.Get(fmt.Sprintf("%s%s", s.URL(), "/v2/keys/foo/bar?wait=true"))
-			body = tests.ReadBodyJSON(resp)
+			watchResp, _ = tests.Get(fmt.Sprintf("%s%s", s.URL(), "/v2/keys/foo/bar?wait=true"))
 			c <- true
 		}()
 
 		// Make sure response didn't fire early.
 		time.Sleep(1 * time.Millisecond)
-		assert.Nil(t, body, "")
 
 		// Set a value.
 		v := url.Values{}
@@ -118,6 +116,7 @@ func TestV2WatchKey(t *testing.T) {
 			t.Fatal("cannot get watch result")
 		}
 
+		body := tests.ReadBodyJSON(watchResp)
 		assert.NotNil(t, body, "")
 		assert.Equal(t, body["action"], "set", "")
 

server/v2/tests/put_handler_test.go

@@ -239,7 +239,7 @@ func TestV2SetKeyCASOnIndexFail(t *testing.T) {
 		body := tests.ReadBodyJSON(resp)
 		assert.Equal(t, body["errorCode"], 101, "")
 		assert.Equal(t, body["message"], "Compare failed", "")
-		assert.Equal(t, body["cause"], "[ != XXX] [10 != 2]", "")
+		assert.Equal(t, body["cause"], "[10 != 2]", "")
 		assert.Equal(t, body["index"], 2, "")
 	})
 }
@@ -307,7 +307,7 @@ func TestV2SetKeyCASOnValueFail(t *testing.T) {
 		body := tests.ReadBodyJSON(resp)
 		assert.Equal(t, body["errorCode"], 101, "")
 		assert.Equal(t, body["message"], "Compare failed", "")
-		assert.Equal(t, body["cause"], "[AAA != XXX] [0 != 2]", "")
+		assert.Equal(t, body["cause"], "[AAA != XXX]", "")
 		assert.Equal(t, body["index"], 2, "")
 	})
 }
@@ -330,6 +330,84 @@ func TestV2SetKeyCASWithMissingValueFails(t *testing.T) {
 	})
 }
 
+// Ensures that a key is not set if both previous value and index do not match.
+//
+//   $ 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 prevValue=AAA -d prevIndex=3
+//
+func TestV2SetKeyCASOnValueAndIndexFail(t *testing.T) {
+	tests.RunServer(func(s *server.Server) {
+		v := url.Values{}
+		v.Set("value", "XXX")
+		fullURL := fmt.Sprintf("%s%s", s.URL(), "/v2/keys/foo/bar")
+		resp, _ := tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusCreated)
+		tests.ReadBody(resp)
+		v.Set("value", "YYY")
+		v.Set("prevValue", "AAA")
+		v.Set("prevIndex", "3")
+		resp, _ = tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusPreconditionFailed)
+		body := tests.ReadBodyJSON(resp)
+		assert.Equal(t, body["errorCode"], 101, "")
+		assert.Equal(t, body["message"], "Compare failed", "")
+		assert.Equal(t, body["cause"], "[AAA != XXX] [3 != 2]", "")
+		assert.Equal(t, body["index"], 2, "")
+	})
+}
+
+// Ensures that a key is not set if previous value match but index does not.
+//
+//   $ 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 prevValue=XXX -d prevIndex=3
+//
+func TestV2SetKeyCASOnValueMatchAndIndexFail(t *testing.T) {
+	tests.RunServer(func(s *server.Server) {
+		v := url.Values{}
+		v.Set("value", "XXX")
+		fullURL := fmt.Sprintf("%s%s", s.URL(), "/v2/keys/foo/bar")
+		resp, _ := tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusCreated)
+		tests.ReadBody(resp)
+		v.Set("value", "YYY")
+		v.Set("prevValue", "XXX")
+		v.Set("prevIndex", "3")
+		resp, _ = tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusPreconditionFailed)
+		body := tests.ReadBodyJSON(resp)
+		assert.Equal(t, body["errorCode"], 101, "")
+		assert.Equal(t, body["message"], "Compare failed", "")
+		assert.Equal(t, body["cause"], "[3 != 2]", "")
+		assert.Equal(t, body["index"], 2, "")
+	})
+}
+
+// Ensures that a key is not set if previous index matches but value does not.
+//
+//   $ 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 prevValue=AAA -d prevIndex=2
+//
+func TestV2SetKeyCASOnIndexMatchAndValueFail(t *testing.T) {
+	tests.RunServer(func(s *server.Server) {
+		v := url.Values{}
+		v.Set("value", "XXX")
+		fullURL := fmt.Sprintf("%s%s", s.URL(), "/v2/keys/foo/bar")
+		resp, _ := tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusCreated)
+		tests.ReadBody(resp)
+		v.Set("value", "YYY")
+		v.Set("prevValue", "AAA")
+		v.Set("prevIndex", "2")
+		resp, _ = tests.PutForm(fullURL, v)
+		assert.Equal(t, resp.StatusCode, http.StatusPreconditionFailed)
+		body := tests.ReadBodyJSON(resp)
+		assert.Equal(t, body["errorCode"], 101, "")
+		assert.Equal(t, body["message"], "Compare failed", "")
+		assert.Equal(t, body["cause"], "[AAA != XXX]", "")
+		assert.Equal(t, body["index"], 2, "")
+	})
+}
+
 // Ensure that we can set an empty value
 //
 //   $ curl -X PUT localhost:4001/v2/keys/foo/bar -d value=