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

Merge pull request #484 from bcwaldon/server-config 

Refactor server init code
release-0.4
Brian Waldon 2014-01-22 11:39:27 -08:00
parent 6c48466bfe 2fe22f1890
commit 394e651591
11 changed files with 397 additions and 423 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
etcd.go
View File

@ -18,15 +18,19 @@ package main
import (
"fmt"
"net"
"net/http"
"os"
"runtime"
"time"
"github.com/coreos/raft"
ehttp "github.com/coreos/etcd/http"
"github.com/coreos/etcd/log"
"github.com/coreos/etcd/metrics"
"github.com/coreos/etcd/server"
"github.com/coreos/etcd/store"
"github.com/coreos/raft"
)
func main() {
@ -98,32 +102,89 @@ func main() {
}
}
// Retrieve CORS configuration
corsInfo, err := ehttp.NewCORSInfo(config.CorsOrigins)
if err != nil {
log.Fatal("CORS:", err)
}
// Create etcd key-value store and registry.
store := store.New()
registry := server.NewRegistry(store)
// Create peer server.
heartbeatTimeout := time.Duration(config.Peer.HeartbeatTimeout) * time.Millisecond
electionTimeout := time.Duration(config.Peer.ElectionTimeout) * time.Millisecond
ps := server.NewPeerServer(info.Name, config.DataDir, info.RaftURL, info.RaftListenHost, &peerTLSConfig, &info.RaftTLS, registry, store, config.SnapshotCount, heartbeatTimeout, electionTimeout, &mb)
ps.MaxClusterSize = config.MaxClusterSize
ps.RetryTimes = config.MaxRetryAttempts
// Create stats objects
followersStats := server.NewRaftFollowersStats(info.Name)
serverStats := server.NewRaftServerStats(info.Name)
// Create client server.
s := server.New(info.Name, info.EtcdURL, info.EtcdListenHost, &tlsConfig, &info.EtcdTLS, ps, registry, store, &mb)
if err := s.AllowOrigins(config.CorsOrigins); err != nil {
// Calculate all of our timeouts
heartbeatTimeout := time.Duration(config.Peer.HeartbeatTimeout) * time.Millisecond
electionTimeout := time.Duration(config.Peer.ElectionTimeout) * time.Millisecond
dialTimeout := (3 * heartbeatTimeout) + electionTimeout
responseHeaderTimeout := (3 * heartbeatTimeout) + electionTimeout
// Create peer server.
psConfig := server.PeerServerConfig{
Name: info.Name,
Scheme: peerTLSConfig.Scheme,
URL: info.RaftURL,
SnapshotCount: config.SnapshotCount,
MaxClusterSize: config.MaxClusterSize,
RetryTimes: config.MaxRetryAttempts,
}
ps := server.NewPeerServer(psConfig, registry, store, &mb, followersStats, serverStats)
var psListener net.Listener
if psConfig.Scheme == "https" {
psListener, err = server.NewTLSListener(info.RaftListenHost, info.RaftTLS.CertFile, info.RaftTLS.KeyFile)
} else {
psListener, err = server.NewListener(info.RaftListenHost)
}
if err != nil {
panic(err)
}
// Create Raft transporter and server
raftTransporter := server.NewTransporter(followersStats, serverStats, registry, heartbeatTimeout, dialTimeout, responseHeaderTimeout)
if psConfig.Scheme == "https" {
raftTransporter.SetTLSConfig(peerTLSConfig.Client)
}
raftServer, err := raft.NewServer(info.Name, config.DataDir, raftTransporter, store, ps, "")
if err != nil {
log.Fatal(err)
}
raftServer.SetElectionTimeout(electionTimeout)
raftServer.SetHeartbeatTimeout(heartbeatTimeout)
ps.SetRaftServer(raftServer)
// Create client server.
s := server.New(info.Name, info.EtcdURL, ps, registry, store, &mb)
if config.Trace() {
s.EnableTracing()
}
var sListener net.Listener
if tlsConfig.Scheme == "https" {
sListener, err = server.NewTLSListener(info.EtcdListenHost, info.EtcdTLS.CertFile, info.EtcdTLS.KeyFile)
} else {
sListener, err = server.NewListener(info.EtcdListenHost)
}
if err != nil {
panic(err)
}
ps.SetServer(s)
ps.Start(config.Snapshot, config.Peers)
// Run peer server in separate thread while the client server blocks.
go func() {
log.Fatal(ps.ListenAndServe(config.Snapshot, config.Peers))
log.Infof("raft server [name %s, listen on %s, advertised url %s]", ps.Config.Name, psListener.Addr(), ps.Config.URL)
sHTTP := &ehttp.CORSHandler{ps.HTTPHandler(), corsInfo}
log.Fatal(http.Serve(psListener, sHTTP))
}()
log.Fatal(s.ListenAndServe())
log.Infof("etcd server [name %s, listen on %s, advertised url %s]", s.Name, sListener.Addr(), s.URL())
sHTTP := &ehttp.CORSHandler{s.HTTPHandler(), corsInfo}
log.Fatal(http.Serve(sListener, sHTTP))
}

37
server/cors_handler.go → http/cors.go
View File

@ -14,56 +14,55 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package server
package http
import (
"fmt"
"net/http"
"net/url"
"github.com/gorilla/mux"
)
type corsHandler struct {
router *mux.Router
corsOrigins map[string]bool
}
type CORSInfo map[string]bool
// AllowOrigins sets a comma-delimited list of origins that are allowed.
func (s *corsHandler) AllowOrigins(origins []string) error {
func NewCORSInfo(origins []string) (*CORSInfo, error) {
// Construct a lookup of all origins.
m := make(map[string]bool)
for _, v := range origins {
if v != "*" {
if _, err := url.Parse(v); err != nil {
return fmt.Errorf("Invalid CORS origin: %s", err)
return nil, fmt.Errorf("Invalid CORS origin: %s", err)
}
}
m[v] = true
}
s.corsOrigins = m
return nil
info := CORSInfo(m)
return &info, nil
}
// OriginAllowed determines whether the server will allow a given CORS origin.
func (c *corsHandler) OriginAllowed(origin string) bool {
return c.corsOrigins["*"] || c.corsOrigins[origin]
func (c CORSInfo) OriginAllowed(origin string) bool {
return c["*"] || c[origin]
}
type CORSHandler struct {
Handler http.Handler
Info *CORSInfo
}
// addHeader adds the correct cors headers given an origin
func (h *corsHandler) addHeader(w http.ResponseWriter, origin string) {
func (h *CORSHandler) addHeader(w http.ResponseWriter, origin string) {
w.Header().Add("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE")
w.Header().Add("Access-Control-Allow-Origin", origin)
}
// ServeHTTP adds the correct CORS headers based on the origin and returns immediatly
// with a 200 OK if the method is OPTIONS.
func (h *corsHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
func (h *CORSHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// Write CORS header.
if h.OriginAllowed("*") {
if h.Info.OriginAllowed("*") {
h.addHeader(w, "*")
} else if origin := req.Header.Get("Origin"); h.OriginAllowed(origin) {
} else if origin := req.Header.Get("Origin"); h.Info.OriginAllowed(origin) {
h.addHeader(w, origin)
}
@ -72,5 +71,5 @@ func (h *corsHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
return
}
h.router.ServeHTTP(w, req)
h.Handler.ServeHTTP(w, req)
}

2
server/join_command.go
View File

@ -52,7 +52,7 @@ func (c *JoinCommand) Apply(context raft.Context) (interface{}, error) {
}
// Check peer number in the cluster
if ps.registry.Count() == ps.MaxClusterSize {
if ps.registry.Count() == ps.Config.MaxClusterSize {
log.Debug("Reject join request from ", c.Name)
return []byte{0}, etcdErr.NewError(etcdErr.EcodeNoMorePeer, "", context.CommitIndex())
}

39
server/listener.go Normal file
View File

@ -0,0 +1,39 @@
package server
import (
"crypto/tls"
"net"
)
func NewListener(addr string) (net.Listener, error) {
if addr == "" {
addr = ":http"
}
l, e := net.Listen("tcp", addr)
if e != nil {
return nil, e
}
return l, nil
}
func NewTLSListener(addr, certFile, keyFile string) (net.Listener, error) {
if addr == "" {
addr = ":https"
}
config := &tls.Config{}
config.NextProtos = []string{"http/1.1"}
var err error
config.Certificates = make([]tls.Certificate, 1)
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return nil, err
}
conn, err := net.Listen("tcp", addr)
if err != nil {
return nil, err
}
return tls.NewListener(conn, config), nil
}

279
server/peer_server.go
View File

@ -2,49 +2,47 @@ package server
import (
"bytes"
"crypto/tls"
"encoding/binary"
"encoding/json"
"fmt"
"io/ioutil"
"net"
"net/http"
"net/url"
"strconv"
"time"
"github.com/coreos/raft"
"github.com/gorilla/mux"
etcdErr "github.com/coreos/etcd/error"
"github.com/coreos/etcd/log"
"github.com/coreos/etcd/metrics"
"github.com/coreos/etcd/store"
"github.com/coreos/raft"
"github.com/gorilla/mux"
)
const retryInterval = 10
const ThresholdMonitorTimeout = 5 * time.Second
type PeerServer struct {
raftServer raft.Server
server *Server
httpServer *http.Server
listener net.Listener
joinIndex uint64
name string
url string
bindAddr string
tlsConf *TLSConfig
tlsInfo *TLSInfo
followersStats *raftFollowersStats
serverStats *raftServerStats
registry *Registry
store store.Store
snapConf *snapshotConf
type PeerServerConfig struct {
Name string
Scheme string
URL string
SnapshotCount int
MaxClusterSize int
RetryTimes int
HeartbeatTimeout time.Duration
ElectionTimeout time.Duration
}
type PeerServer struct {
Config PeerServerConfig
raftServer raft.Server
server *Server
joinIndex uint64
followersStats *raftFollowersStats
serverStats *raftServerStats
registry *Registry
store store.Store
snapConf *snapshotConf
closeChan chan bool
timeoutThresholdChan chan interface{}
@ -65,84 +63,56 @@ type snapshotConf struct {
snapshotThr uint64
}
func NewPeerServer(name string, path string, url string, bindAddr string, tlsConf *TLSConfig, tlsInfo *TLSInfo, registry *Registry, store store.Store, snapshotCount int, heartbeatTimeout, electionTimeout time.Duration, mb *metrics.Bucket) *PeerServer {
func NewPeerServer(psConfig PeerServerConfig, registry *Registry, store store.Store, mb *metrics.Bucket, followersStats *raftFollowersStats, serverStats *raftServerStats) *PeerServer {
s := &PeerServer{
name: name,
url: url,
bindAddr: bindAddr,
tlsConf: tlsConf,
tlsInfo: tlsInfo,
Config: psConfig,
registry: registry,
store: store,
followersStats: &raftFollowersStats{
Leader: name,
Followers: make(map[string]*raftFollowerStats),
},
serverStats: &raftServerStats{
Name: name,
StartTime: time.Now(),
sendRateQueue: &statsQueue{
back: -1,
},
recvRateQueue: &statsQueue{
back: -1,
},
},
HeartbeatTimeout: heartbeatTimeout,
ElectionTimeout: electionTimeout,
followersStats: followersStats,
serverStats: serverStats,
timeoutThresholdChan: make(chan interface{}, 1),
metrics: mb,
}
// Create transporter for raft
raftTransporter := newTransporter(tlsConf.Scheme, tlsConf.Client, s)
// Create raft server
raftServer, err := raft.NewServer(name, path, raftTransporter, s.store, s, "")
if err != nil {
log.Fatal(err)
}
return s
}
func (s *PeerServer) SetRaftServer(raftServer raft.Server) {
s.snapConf = &snapshotConf{
checkingInterval: time.Second * 3,
// this is not accurate, we will update raft to provide an api
lastIndex: raftServer.CommitIndex(),
snapshotThr: uint64(snapshotCount),
snapshotThr: uint64(s.Config.SnapshotCount),
}
raftServer.AddEventListener(raft.StateChangeEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.LeaderChangeEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.TermChangeEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.AddPeerEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.RemovePeerEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.HeartbeatTimeoutEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.ElectionTimeoutThresholdEventType, s.raftEventLogger)
raftServer.AddEventListener(raft.HeartbeatEventType, s.recordMetricEvent)
s.raftServer = raftServer
s.raftServer.AddEventListener(raft.StateChangeEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.LeaderChangeEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.TermChangeEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.AddPeerEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.RemovePeerEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.HeartbeatTimeoutEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.ElectionTimeoutThresholdEventType, s.raftEventLogger)
s.raftServer.AddEventListener(raft.HeartbeatEventType, s.recordMetricEvent)
return s
}
// Start the raft server
func (s *PeerServer) ListenAndServe(snapshot bool, cluster []string) error {
func (s *PeerServer) Start(snapshot bool, cluster []string) error {
// LoadSnapshot
if snapshot {
err := s.raftServer.LoadSnapshot()
if err == nil {
log.Debugf("%s finished load snapshot", s.name)
log.Debugf("%s finished load snapshot", s.Config.Name)
} else {
log.Debug(err)
}
}
s.raftServer.SetElectionTimeout(s.ElectionTimeout)
s.raftServer.SetHeartbeatTimeout(s.HeartbeatTimeout)
s.raftServer.Start()
if s.raftServer.IsLogEmpty() {
@ -155,7 +125,7 @@ func (s *PeerServer) ListenAndServe(snapshot bool, cluster []string) error {
} else {
// Rejoin the previous cluster
cluster = s.registry.PeerURLs(s.raftServer.Leader(), s.name)
cluster = s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name)
for i := 0; i < len(cluster); i++ {
u, err := url.Parse(cluster[i])
if err != nil {
@ -168,7 +138,7 @@ func (s *PeerServer) ListenAndServe(snapshot bool, cluster []string) error {
log.Warn("the entire cluster is down! this peer will restart the cluster.")
}
log.Debugf("%s restart as a follower", s.name)
log.Debugf("%s restart as a follower", s.Config.Name)
}
s.closeChan = make(chan bool)
@ -181,114 +151,19 @@ func (s *PeerServer) ListenAndServe(snapshot bool, cluster []string) error {
go s.monitorSnapshot()
}
// start to response to raft requests
return s.startTransport(s.tlsConf.Scheme, s.tlsConf.Server)
return nil
}
// Overridden version of net/http added so we can manage the listener.
func (s *PeerServer) listenAndServe() error {
addr := s.httpServer.Addr
if addr == "" {
addr = ":http"
}
l, e := net.Listen("tcp", addr)
if e != nil {
return e
}
s.listener = l
return s.httpServer.Serve(l)
}
// Overridden version of net/http added so we can manage the listener.
func (s *PeerServer) listenAndServeTLS(certFile, keyFile string) error {
addr := s.httpServer.Addr
if addr == "" {
addr = ":https"
}
config := &tls.Config{}
if s.httpServer.TLSConfig != nil {
*config = *s.httpServer.TLSConfig
}
if config.NextProtos == nil {
config.NextProtos = []string{"http/1.1"}
}
var err error
config.Certificates = make([]tls.Certificate, 1)
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return err
}
conn, err := net.Listen("tcp", addr)
if err != nil {
return err
}
tlsListener := tls.NewListener(conn, config)
s.listener = tlsListener
return s.httpServer.Serve(tlsListener)
}
// Stops the server.
func (s *PeerServer) Close() {
func (s *PeerServer) Stop() {
if s.closeChan != nil {
close(s.closeChan)
s.closeChan = nil
}
if s.listener != nil {
s.listener.Close()
s.listener = nil
}
}
// Retrieves the underlying Raft server.
func (s *PeerServer) RaftServer() raft.Server {
return s.raftServer
}
// Associates the client server with the peer server.
func (s *PeerServer) SetServer(server *Server) {
s.server = server
}
func (s *PeerServer) startAsLeader() {
// leader need to join self as a peer
for {
_, err := s.raftServer.Do(NewJoinCommand(store.MinVersion(), store.MaxVersion(), s.raftServer.Name(), s.url, s.server.URL()))
if err == nil {
break
}
}
log.Debugf("%s start as a leader", s.name)
}
func (s *PeerServer) startAsFollower(cluster []string) {
// start as a follower in a existing cluster
for i := 0; i < s.RetryTimes; i++ {
ok := s.joinCluster(cluster)
if ok {
return
}
log.Warnf("cannot join to cluster via given peers, retry in %d seconds", retryInterval)
time.Sleep(time.Second * retryInterval)
}
log.Fatalf("Cannot join the cluster via given peers after %x retries", s.RetryTimes)
}
// Start to listen and response raft command
func (s *PeerServer) startTransport(scheme string, tlsConf tls.Config) error {
log.Infof("raft server [name %s, listen on %s, advertised url %s]", s.name, s.bindAddr, s.url)
func (s *PeerServer) HTTPHandler() http.Handler {
router := mux.NewRouter()
s.httpServer = &http.Server{
Handler: router,
TLSConfig: &tlsConf,
Addr: s.bindAddr,
}
// internal commands
router.HandleFunc("/name", s.NameHttpHandler)
router.HandleFunc("/version", s.VersionHttpHandler)
@ -303,12 +178,42 @@ func (s *PeerServer) startTransport(scheme string, tlsConf tls.Config) error {
router.HandleFunc("/snapshotRecovery", s.SnapshotRecoveryHttpHandler)
router.HandleFunc("/etcdURL", s.EtcdURLHttpHandler)
if scheme == "http" {
return s.listenAndServe()
} else {
return s.listenAndServeTLS(s.tlsInfo.CertFile, s.tlsInfo.KeyFile)
return router
}
// Retrieves the underlying Raft server.
func (s *PeerServer) RaftServer() raft.Server {
return s.raftServer
}
// Associates the client server with the peer server.
func (s *PeerServer) SetServer(server *Server) {
s.server = server
}
func (s *PeerServer) startAsLeader() {
// leader need to join self as a peer
for {
_, err := s.raftServer.Do(NewJoinCommand(store.MinVersion(), store.MaxVersion(), s.raftServer.Name(), s.Config.URL, s.server.URL()))
if err == nil {
break
}
}
log.Debugf("%s start as a leader", s.Config.Name)
}
func (s *PeerServer) startAsFollower(cluster []string) {
// start as a follower in a existing cluster
for i := 0; i < s.Config.RetryTimes; i++ {
ok := s.joinCluster(cluster)
if ok {
return
}
log.Warnf("cannot join to cluster via given peers, retry in %d seconds", retryInterval)
time.Sleep(time.Second * retryInterval)
}
log.Fatalf("Cannot join the cluster via given peers after %x retries", s.Config.RetryTimes)
}
// getVersion fetches the peer version of a cluster.
@ -333,14 +238,14 @@ func getVersion(t *transporter, versionURL url.URL) (int, error) {
// Upgradable checks whether all peers in a cluster support an upgrade to the next store version.
func (s *PeerServer) Upgradable() error {
nextVersion := s.store.Version() + 1
for _, peerURL := range s.registry.PeerURLs(s.raftServer.Leader(), s.name) {
for _, peerURL := range s.registry.PeerURLs(s.raftServer.Leader(), s.Config.Name) {
u, err := url.Parse(peerURL)
if err != nil {
return fmt.Errorf("PeerServer: Cannot parse URL: '%s' (%s)", peerURL, err)
}
t, _ := s.raftServer.Transporter().(*transporter)
checkURL := (&url.URL{Host: u.Host, Scheme: s.tlsConf.Scheme, Path: fmt.Sprintf("/version/%d/check", nextVersion)}).String()
checkURL := (&url.URL{Host: u.Host, Scheme: s.Config.Scheme, Path: fmt.Sprintf("/version/%d/check", nextVersion)}).String()
resp, _, err := t.Get(checkURL)
if err != nil {
return fmt.Errorf("PeerServer: Cannot check version compatibility: %s", u.Host)
@ -359,9 +264,9 @@ func (s *PeerServer) joinCluster(cluster []string) bool {
continue
}
err := s.joinByPeer(s.raftServer, peer, s.tlsConf.Scheme)
err := s.joinByPeer(s.raftServer, peer, s.Config.Scheme)
if err == nil {
log.Debugf("%s success join to the cluster via peer %s", s.name, peer)
log.Debugf("%s success join to the cluster via peer %s", s.Config.Name, peer)
return true
} else {
@ -392,7 +297,7 @@ func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string)
return fmt.Errorf("Unable to join: cluster version is %d; version compatibility is %d - %d", version, store.MinVersion(), store.MaxVersion())
}
json.NewEncoder(&b).Encode(NewJoinCommand(store.MinVersion(), store.MaxVersion(), server.Name(), s.url, s.server.URL()))
json.NewEncoder(&b).Encode(NewJoinCommand(store.MinVersion(), store.MaxVersion(), server.Name(), s.Config.URL, s.server.URL()))
joinURL := url.URL{Host: peer, Scheme: scheme, Path: "/join"}
@ -417,7 +322,7 @@ func (s *PeerServer) joinByPeer(server raft.Server, peer string, scheme string)
if resp.StatusCode == http.StatusTemporaryRedirect {
address := resp.Header.Get("Location")
log.Debugf("Send Join Request to %s", address)
json.NewEncoder(&b).Encode(NewJoinCommand(store.MinVersion(), store.MaxVersion(), server.Name(), s.url, s.server.URL()))
json.NewEncoder(&b).Encode(NewJoinCommand(store.MinVersion(), store.MaxVersion(), server.Name(), s.Config.URL, s.server.URL()))
resp, req, err = t.Post(address, &b)
} else if resp.StatusCode == http.StatusBadRequest {
@ -477,21 +382,21 @@ func (s *PeerServer) raftEventLogger(event raft.Event) {
switch event.Type() {
case raft.StateChangeEventType:
log.Infof("%s: state changed from '%v' to '%v'.", s.name, prevValue, value)
log.Infof("%s: state changed from '%v' to '%v'.", s.Config.Name, prevValue, value)
case raft.TermChangeEventType:
log.Infof("%s: term #%v started.", s.name, value)
log.Infof("%s: term #%v started.", s.Config.Name, value)
case raft.LeaderChangeEventType:
log.Infof("%s: leader changed from '%v' to '%v'.", s.name, prevValue, value)
log.Infof("%s: leader changed from '%v' to '%v'.", s.Config.Name, prevValue, value)
case raft.AddPeerEventType:
log.Infof("%s: peer added: '%v'", s.name, value)
log.Infof("%s: peer added: '%v'", s.Config.Name, value)
case raft.RemovePeerEventType:
log.Infof("%s: peer removed: '%v'", s.name, value)
log.Infof("%s: peer removed: '%v'", s.Config.Name, value)
case raft.HeartbeatTimeoutEventType:
var name = "<unknown>"
if peer, ok := value.(*raft.Peer); ok {
name = peer.Name
}
log.Infof("%s: warning: heartbeat timed out: '%v'", s.name, name)
log.Infof("%s: warning: heartbeat timed out: '%v'", s.Config.Name, name)
case raft.ElectionTimeoutThresholdEventType:
select {
case s.timeoutThresholdChan <- value:
@ -538,7 +443,7 @@ func (s *PeerServer) monitorTimeoutThreshold(closeChan chan bool) {
for {
select {
case value := <-s.timeoutThresholdChan:
log.Infof("%s: warning: heartbeat near election timeout: %v", s.name, value)
log.Infof("%s: warning: heartbeat near election timeout: %v", s.Config.Name, value)
case <-closeChan:
return
}

37
server/peer_server_handlers.go
View File

@ -15,7 +15,7 @@ import (
// Get all the current logs
func (ps *PeerServer) GetLogHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] GET %s/log", ps.url)
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())
@ -27,11 +27,11 @@ func (ps *PeerServer) VoteHttpHandler(w http.ResponseWriter, req *http.Request)
if _, err := rvreq.Decode(req.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
log.Warnf("[recv] BADREQUEST %s/vote [%v]", ps.url, err)
log.Warnf("[recv] BADREQUEST %s/vote [%v]", ps.Config.URL, err)
return
}
log.Debugf("[recv] POST %s/vote [%s]", ps.url, rvreq.CandidateName)
log.Debugf("[recv] POST %s/vote [%s]", ps.Config.URL, rvreq.CandidateName)
resp := ps.raftServer.RequestVote(rvreq)
@ -55,11 +55,11 @@ func (ps *PeerServer) AppendEntriesHttpHandler(w http.ResponseWriter, req *http.
if _, err := aereq.Decode(req.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
log.Warnf("[recv] BADREQUEST %s/log/append [%v]", ps.url, err)
log.Warnf("[recv] BADREQUEST %s/log/append [%v]", ps.Config.URL, err)
return
}
log.Debugf("[recv] POST %s/log/append [%d]", ps.url, len(aereq.Entries))
log.Debugf("[recv] POST %s/log/append [%d]", ps.Config.URL, len(aereq.Entries))
ps.serverStats.RecvAppendReq(aereq.LeaderName, int(req.ContentLength))
@ -90,11 +90,11 @@ func (ps *PeerServer) SnapshotHttpHandler(w http.ResponseWriter, req *http.Reque
if _, err := ssreq.Decode(req.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
log.Warnf("[recv] BADREQUEST %s/snapshot [%v]", ps.url, err)
log.Warnf("[recv] BADREQUEST %s/snapshot [%v]", ps.Config.URL, err)
return
}
log.Debugf("[recv] POST %s/snapshot", ps.url)
log.Debugf("[recv] POST %s/snapshot", ps.Config.URL)
resp := ps.raftServer.RequestSnapshot(ssreq)
@ -117,11 +117,11 @@ func (ps *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *ht
if _, err := ssrreq.Decode(req.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
log.Warnf("[recv] BADREQUEST %s/snapshotRecovery [%v]", ps.url, err)
log.Warnf("[recv] BADREQUEST %s/snapshotRecovery [%v]", ps.Config.URL, err)
return
}
log.Debugf("[recv] POST %s/snapshotRecovery", ps.url)
log.Debugf("[recv] POST %s/snapshotRecovery", ps.Config.URL)
resp := ps.raftServer.SnapshotRecoveryRequest(ssrreq)
@ -140,7 +140,7 @@ func (ps *PeerServer) SnapshotRecoveryHttpHandler(w http.ResponseWriter, req *ht
// 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.url)
log.Debugf("[recv] Get %s/etcdURL/ ", ps.Config.URL)
w.WriteHeader(http.StatusOK)
w.Write([]byte(ps.server.URL()))
}
@ -149,13 +149,6 @@ func (ps *PeerServer) EtcdURLHttpHandler(w http.ResponseWriter, req *http.Reques
func (ps *PeerServer) JoinHttpHandler(w http.ResponseWriter, req *http.Request) {
command := &JoinCommand{}
// Write CORS header.
if ps.server.OriginAllowed("*") {
w.Header().Add("Access-Control-Allow-Origin", "*")
} else if ps.server.OriginAllowed(req.Header.Get("Origin")) {
w.Header().Add("Access-Control-Allow-Origin", req.Header.Get("Origin"))
}
err := decodeJsonRequest(req, command)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
@ -195,21 +188,21 @@ func (ps *PeerServer) RemoveHttpHandler(w http.ResponseWriter, req *http.Request
// Response to the name request
func (ps *PeerServer) NameHttpHandler(w http.ResponseWriter, req *http.Request) {
log.Debugf("[recv] Get %s/name/ ", ps.url)
log.Debugf("[recv] Get %s/name/ ", ps.Config.URL)
w.WriteHeader(http.StatusOK)
w.Write([]byte(ps.name))
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.url)
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.url, req.URL.Path)
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() {
@ -221,7 +214,7 @@ func (ps *PeerServer) VersionCheckHttpHandler(w http.ResponseWriter, req *http.R
// 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.url)
log.Debugf("[recv] Get %s/version", ps.Config.URL)
// Check if upgrade is possible for all nodes.
if err := ps.Upgradable(); err != nil {

7
server/raft_follower_stats.go
View File

@ -10,6 +10,13 @@ type raftFollowersStats struct {
Followers map[string]*raftFollowerStats `json:"followers"`
}
func NewRaftFollowersStats(name string) *raftFollowersStats {
return &raftFollowersStats{
Leader: name,
Followers: make(map[string]*raftFollowerStats),
}
}
type raftFollowerStats struct {
Latency struct {
Current float64 `json:"current"`

13
server/raft_server_stats.go
View File

@ -29,6 +29,19 @@ type raftServerStats struct {
recvRateQueue *statsQueue
}
func NewRaftServerStats(name string) *raftServerStats {
return &raftServerStats{
Name: name,
StartTime: time.Now(),
sendRateQueue: &statsQueue{
back: -1,
},
recvRateQueue: &statsQueue{
back: -1,
},
}
}
func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) {
ss.State = raft.Follower
if leaderName != ss.LeaderInfo.Name {

218
server/server.go
View File

@ -1,15 +1,16 @@
package server
import (
"crypto/tls"
"encoding/json"
"fmt"
"net"
"net/http"
"net/http/pprof"
"strings"
"time"
"github.com/coreos/raft"
"github.com/gorilla/mux"
etcdErr "github.com/coreos/etcd/error"
"github.com/coreos/etcd/log"
"github.com/coreos/etcd/metrics"
@ -18,60 +19,37 @@ import (
"github.com/coreos/etcd/server/v2"
"github.com/coreos/etcd/store"
_ "github.com/coreos/etcd/store/v2"
"github.com/coreos/raft"
"github.com/gorilla/mux"
)
// This is the default implementation of the Server interface.
type Server struct {
http.Server
peerServer *PeerServer
registry *Registry
listener net.Listener
store store.Store
name string
url string
tlsConf *TLSConfig
tlsInfo *TLSInfo
router *mux.Router
corsHandler *corsHandler
metrics *metrics.Bucket
Name string
url string
handler http.Handler
peerServer *PeerServer
registry *Registry
store store.Store
metrics *metrics.Bucket
trace bool
}
// Creates a new Server.
func New(name string, urlStr string, bindAddr string, tlsConf *TLSConfig, tlsInfo *TLSInfo, peerServer *PeerServer, registry *Registry, store store.Store, mb *metrics.Bucket) *Server {
r := mux.NewRouter()
cors := &corsHandler{router: r}
func New(name, url string, peerServer *PeerServer, registry *Registry, store store.Store, mb *metrics.Bucket) *Server {
s := &Server{
Server: http.Server{
Handler: cors,
TLSConfig: &tlsConf.Server,
Addr: bindAddr,
},
name: name,
store: store,
registry: registry,
url: urlStr,
tlsConf: tlsConf,
tlsInfo: tlsInfo,
peerServer: peerServer,
router: r,
corsHandler: cors,
Name: name,
url: url,
store: store,
registry: registry,
peerServer: peerServer,
metrics: mb,
}
// Install the routes.
s.handleFunc("/version", s.GetVersionHandler).Methods("GET")
s.installV1()
s.installV2()
s.installMod()
return s
}
func (s *Server) EnableTracing() {
s.installDebug()
s.trace = true
}
// The current state of the server in the cluster.
@ -114,69 +92,67 @@ func (s *Server) Store() store.Store {
return s.store
}
func (s *Server) installV1() {
s.handleFuncV1("/v1/keys/{key:.*}", v1.GetKeyHandler).Methods("GET")
s.handleFuncV1("/v1/keys/{key:.*}", v1.SetKeyHandler).Methods("POST", "PUT")
s.handleFuncV1("/v1/keys/{key:.*}", v1.DeleteKeyHandler).Methods("DELETE")
s.handleFuncV1("/v1/watch/{key:.*}", v1.WatchKeyHandler).Methods("GET", "POST")
s.handleFunc("/v1/leader", s.GetLeaderHandler).Methods("GET")
s.handleFunc("/v1/machines", s.GetPeersHandler).Methods("GET")
s.handleFunc("/v1/peers", s.GetPeersHandler).Methods("GET")
s.handleFunc("/v1/stats/self", s.GetStatsHandler).Methods("GET")
s.handleFunc("/v1/stats/leader", s.GetLeaderStatsHandler).Methods("GET")
s.handleFunc("/v1/stats/store", s.GetStoreStatsHandler).Methods("GET")
func (s *Server) installV1(r *mux.Router) {
s.handleFuncV1(r, "/v1/keys/{key:.*}", v1.GetKeyHandler).Methods("GET")
s.handleFuncV1(r, "/v1/keys/{key:.*}", v1.SetKeyHandler).Methods("POST", "PUT")
s.handleFuncV1(r, "/v1/keys/{key:.*}", v1.DeleteKeyHandler).Methods("DELETE")
s.handleFuncV1(r, "/v1/watch/{key:.*}", v1.WatchKeyHandler).Methods("GET", "POST")
s.handleFunc(r, "/v1/leader", s.GetLeaderHandler).Methods("GET")
s.handleFunc(r, "/v1/machines", s.GetPeersHandler).Methods("GET")
s.handleFunc(r, "/v1/peers", s.GetPeersHandler).Methods("GET")
s.handleFunc(r, "/v1/stats/self", s.GetStatsHandler).Methods("GET")
s.handleFunc(r, "/v1/stats/leader", s.GetLeaderStatsHandler).Methods("GET")
s.handleFunc(r, "/v1/stats/store", s.GetStoreStatsHandler).Methods("GET")
}
func (s *Server) installV2() {
s.handleFuncV2("/v2/keys/{key:.*}", v2.GetHandler).Methods("GET")
s.handleFuncV2("/v2/keys/{key:.*}", v2.PostHandler).Methods("POST")
s.handleFuncV2("/v2/keys/{key:.*}", v2.PutHandler).Methods("PUT")
s.handleFuncV2("/v2/keys/{key:.*}", v2.DeleteHandler).Methods("DELETE")
s.handleFunc("/v2/leader", s.GetLeaderHandler).Methods("GET")
s.handleFunc("/v2/machines", s.GetPeersHandler).Methods("GET")
s.handleFunc("/v2/peers", s.GetPeersHandler).Methods("GET")
s.handleFunc("/v2/stats/self", s.GetStatsHandler).Methods("GET")
s.handleFunc("/v2/stats/leader", s.GetLeaderStatsHandler).Methods("GET")
s.handleFunc("/v2/stats/store", s.GetStoreStatsHandler).Methods("GET")
s.handleFunc("/v2/speedTest", s.SpeedTestHandler).Methods("GET")
func (s *Server) installV2(r *mux.Router) {
s.handleFuncV2(r, "/v2/keys/{key:.*}", v2.GetHandler).Methods("GET")
s.handleFuncV2(r, "/v2/keys/{key:.*}", v2.PostHandler).Methods("POST")
s.handleFuncV2(r, "/v2/keys/{key:.*}", v2.PutHandler).Methods("PUT")
s.handleFuncV2(r, "/v2/keys/{key:.*}", v2.DeleteHandler).Methods("DELETE")
s.handleFunc(r, "/v2/leader", s.GetLeaderHandler).Methods("GET")
s.handleFunc(r, "/v2/machines", s.GetPeersHandler).Methods("GET")
s.handleFunc(r, "/v2/peers", s.GetPeersHandler).Methods("GET")
s.handleFunc(r, "/v2/stats/self", s.GetStatsHandler).Methods("GET")
s.handleFunc(r, "/v2/stats/leader", s.GetLeaderStatsHandler).Methods("GET")
s.handleFunc(r, "/v2/stats/store", s.GetStoreStatsHandler).Methods("GET")
s.handleFunc(r, "/v2/speedTest", s.SpeedTestHandler).Methods("GET")
}
func (s *Server) installMod() {
r := s.router
r.PathPrefix("/mod").Handler(http.StripPrefix("/mod", mod.HttpHandler(s.url)))
func (s *Server) installMod(r *mux.Router) {
r.PathPrefix("/mod").Handler(http.StripPrefix("/mod", mod.HttpHandler(s.URL())))
}
func (s *Server) installDebug() {
s.handleFunc("/debug/metrics", s.GetMetricsHandler).Methods("GET")
s.router.HandleFunc("/debug/pprof", pprof.Index)
s.router.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
s.router.HandleFunc("/debug/pprof/profile", pprof.Profile)
s.router.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
s.router.HandleFunc("/debug/pprof/{name}", pprof.Index)
func (s *Server) installDebug(r *mux.Router) {
s.handleFunc(r, "/debug/metrics", s.GetMetricsHandler).Methods("GET")
r.HandleFunc("/debug/pprof", pprof.Index)
r.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
r.HandleFunc("/debug/pprof/profile", pprof.Profile)
r.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
r.HandleFunc("/debug/pprof/{name}", pprof.Index)
}
// Adds a v1 server handler to the router.
func (s *Server) handleFuncV1(path string, f func(http.ResponseWriter, *http.Request, v1.Server) error) *mux.Route {
return s.handleFunc(path, func(w http.ResponseWriter, req *http.Request) error {
func (s *Server) handleFuncV1(r *mux.Router, path string, f func(http.ResponseWriter, *http.Request, v1.Server) error) *mux.Route {
return s.handleFunc(r, path, func(w http.ResponseWriter, req *http.Request) error {
return f(w, req, s)
})
}
// Adds a v2 server handler to the router.
func (s *Server) handleFuncV2(path string, f func(http.ResponseWriter, *http.Request, v2.Server) error) *mux.Route {
return s.handleFunc(path, func(w http.ResponseWriter, req *http.Request) error {
func (s *Server) handleFuncV2(r *mux.Router, path string, f func(http.ResponseWriter, *http.Request, v2.Server) error) *mux.Route {
return s.handleFunc(r, path, func(w http.ResponseWriter, req *http.Request) error {
return f(w, req, s)
})
}
// Adds a server handler to the router.
func (s *Server) handleFunc(path string, f func(http.ResponseWriter, *http.Request) error) *mux.Route {
r := s.router
func (s *Server) handleFunc(r *mux.Router, path string, f func(http.ResponseWriter, *http.Request) error) *mux.Route {
// Wrap the standard HandleFunc interface to pass in the server reference.
return r.HandleFunc(path, func(w http.ResponseWriter, req *http.Request) {
// Log request.
log.Debugf("[recv] %s %s %s [%s]", req.Method, s.url, req.URL.Path, req.RemoteAddr)
log.Debugf("[recv] %s %s %s [%s]", req.Method, s.URL(), req.URL.Path, req.RemoteAddr)
// Execute handler function and return error if necessary.
if err := f(w, req); err != nil {
@ -191,68 +167,20 @@ func (s *Server) handleFunc(path string, f func(http.ResponseWriter, *http.Reque
})
}
// Start to listen and response etcd client command
func (s *Server) ListenAndServe() error {
log.Infof("etcd server [name %s, listen on %s, advertised url %s]", s.name, s.Server.Addr, s.url)
func (s *Server) HTTPHandler() http.Handler {
router := mux.NewRouter()
if s.tlsConf.Scheme == "http" {
return s.listenAndServe()
} else {
return s.listenAndServeTLS(s.tlsInfo.CertFile, s.tlsInfo.KeyFile)
}
}
// Install the routes.
s.handleFunc(router, "/version", s.GetVersionHandler).Methods("GET")
s.installV1(router)
s.installV2(router)
s.installMod(router)
// Overridden version of net/http added so we can manage the listener.
func (s *Server) listenAndServe() error {
addr := s.Server.Addr
if addr == "" {
addr = ":http"
}
l, e := net.Listen("tcp", addr)
if e != nil {
return e
}
s.listener = l
return s.Server.Serve(l)
}
// Overridden version of net/http added so we can manage the listener.
func (s *Server) listenAndServeTLS(certFile, keyFile string) error {
addr := s.Server.Addr
if addr == "" {
addr = ":https"
}
config := &tls.Config{}
if s.Server.TLSConfig != nil {
*config = *s.Server.TLSConfig
}
if config.NextProtos == nil {
config.NextProtos = []string{"http/1.1"}
if s.trace {
s.installDebug(router)
}
var err error
config.Certificates = make([]tls.Certificate, 1)
config.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return err
}
conn, err := net.Listen("tcp", addr)
if err != nil {
return err
}
tlsListener := tls.NewListener(conn, config)
s.listener = tlsListener
return s.Server.Serve(tlsListener)
}
// Stops the server.
func (s *Server) Close() {
if s.listener != nil {
s.listener.Close()
s.listener = nil
}
return router
}
// Dispatch command to the current leader
@ -322,16 +250,6 @@ func (s *Server) Dispatch(c raft.Command, w http.ResponseWriter, req *http.Reque
}
}
// OriginAllowed determines whether the server will allow a given CORS origin.
func (s *Server) OriginAllowed(origin string) bool {
return s.corsHandler.OriginAllowed(origin)
}
// AllowOrigins sets a comma-delimited list of origins that are allowed.
func (s *Server) AllowOrigins(origins []string) error {
return s.corsHandler.AllowOrigins(origins)
}
// Handler to return the current version of etcd.
func (s *Server) GetVersionHandler(w http.ResponseWriter, req *http.Request) error {
w.WriteHeader(http.StatusOK)
@ -353,7 +271,7 @@ func (s *Server) GetLeaderHandler(w http.ResponseWriter, req *http.Request) erro
// Handler to return all the known peers in the current cluster.
func (s *Server) GetPeersHandler(w http.ResponseWriter, req *http.Request) error {
peers := s.registry.ClientURLs(s.peerServer.RaftServer().Leader(), s.name)
peers := s.registry.ClientURLs(s.peerServer.RaftServer().Leader(), s.Name)
w.WriteHeader(http.StatusOK)
w.Write([]byte(strings.Join(peers, ", ")))
return nil

53
server/transporter.go
View File

@ -15,9 +15,13 @@ import (
// Transporter layer for communication between raft nodes
type transporter struct {
client *http.Client
transport *http.Transport
peerServer *PeerServer
requestTimeout time.Duration
followersStats *raftFollowersStats
serverStats *raftServerStats
registry *Registry
client *http.Client
transport *http.Transport
}
type dialer func(network, addr string) (net.Conn, error)
@ -25,13 +29,7 @@ type dialer func(network, addr string) (net.Conn, error)
// Create transporter using by raft server
// Create http or https transporter based on
// whether the user give the server cert and key
func newTransporter(scheme string, tlsConf tls.Config, peerServer *PeerServer) *transporter {
// names for each type of timeout, for the sake of clarity
dialTimeout := (3 * peerServer.HeartbeatTimeout) + peerServer.ElectionTimeout
responseHeaderTimeout := (3 * peerServer.HeartbeatTimeout) + peerServer.ElectionTimeout
t := transporter{}
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)
@ -39,18 +37,23 @@ func newTransporter(scheme string, tlsConf tls.Config, peerServer *PeerServer) *
ResponseHeaderTimeout: responseHeaderTimeout,
}
if scheme == "https" {
tr.TLSClientConfig = &tlsConf
tr.DisableCompression = true
t := transporter{
client: &http.Client{Transport: tr},
transport: tr,
requestTimeout: requestTimeout,
followersStats: followersStats,
serverStats: serverStats,
registry: registry,
}
t.client = &http.Client{Transport: tr}
t.transport = tr
t.peerServer = peerServer
return &t
}
func (t *transporter) SetTLSConfig(tlsConf tls.Config) {
t.transport.TLSClientConfig = &tlsConf
t.transport.DisableCompression = true
}
// Sends AppendEntries RPCs to a peer when the server is the leader.
func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Peer, req *raft.AppendEntriesRequest) *raft.AppendEntriesResponse {
var b bytes.Buffer
@ -62,18 +65,18 @@ func (t *transporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Pe
size := b.Len()
t.peerServer.serverStats.SendAppendReq(size)
t.serverStats.SendAppendReq(size)
u, _ := t.peerServer.registry.PeerURL(peer.Name)
u, _ := t.registry.PeerURL(peer.Name)
log.Debugf("Send LogEntries to %s ", u)
thisFollowerStats, ok := t.peerServer.followersStats.Followers[peer.Name]
thisFollowerStats, ok := t.followersStats.Followers[peer.Name]
if !ok { //this is the first time this follower has been seen
thisFollowerStats = &raftFollowerStats{}
thisFollowerStats.Latency.Minimum = 1 << 63
t.peerServer.followersStats.Followers[peer.Name] = thisFollowerStats
t.followersStats.Followers[peer.Name] = thisFollowerStats
}
start := time.Now()
@ -119,7 +122,7 @@ func (t *transporter) SendVoteRequest(server raft.Server, peer *raft.Peer, req *
return nil
}
u, _ := t.peerServer.registry.PeerURL(peer.Name)
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)
@ -152,7 +155,7 @@ func (t *transporter) SendSnapshotRequest(server raft.Server, peer *raft.Peer, r
return nil
}
u, _ := t.peerServer.registry.PeerURL(peer.Name)
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)
@ -185,7 +188,7 @@ func (t *transporter) SendSnapshotRecoveryRequest(server raft.Server, peer *raft
return nil
}
u, _ := t.peerServer.registry.PeerURL(peer.Name)
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)
@ -227,7 +230,7 @@ func (t *transporter) Get(urlStr string) (*http.Response, *http.Request, error)
// Cancel the on fly HTTP transaction when timeout happens.
func (t *transporter) CancelWhenTimeout(req *http.Request) {
go func() {
time.Sleep(t.peerServer.HeartbeatTimeout)
time.Sleep(t.requestTimeout)
t.transport.CancelRequest(req)
}()
}

50
tests/server_utils.go
View File

@ -2,9 +2,12 @@ package tests
import (
"io/ioutil"
"net/http"
"os"
"time"
"github.com/coreos/raft"
"github.com/coreos/etcd/server"
"github.com/coreos/etcd/store"
)
@ -26,23 +29,55 @@ func RunServer(f func(*server.Server)) {
store := store.New()
registry := server.NewRegistry(store)
ps := server.NewPeerServer(testName, path, "http://"+testRaftURL, testRaftURL, &server.TLSConfig{Scheme: "http"}, &server.TLSInfo{}, registry, store, testSnapshotCount, testHeartbeatTimeout, testElectionTimeout, nil)
ps.MaxClusterSize = 9
s := server.New(testName, "http://"+testClientURL, testClientURL, &server.TLSConfig{Scheme: "http"}, &server.TLSInfo{}, ps, registry, store, nil)
serverStats := server.NewRaftServerStats(testName)
followersStats := server.NewRaftFollowersStats(testName)
psConfig := server.PeerServerConfig{
Name: testName,
URL: "http://"+testRaftURL,
Scheme: "http",
SnapshotCount: testSnapshotCount,
MaxClusterSize: 9,
}
ps := server.NewPeerServer(psConfig, registry, store, nil, followersStats, serverStats)
psListener, err := server.NewListener(testRaftURL)
if err != nil {
panic(err)
}
// Create Raft transporter and server
dialTimeout := (3 * testHeartbeatTimeout) + testElectionTimeout
responseHeaderTimeout := (3 * testHeartbeatTimeout) + testElectionTimeout
raftTransporter := server.NewTransporter(followersStats, serverStats, registry, testHeartbeatTimeout, dialTimeout, responseHeaderTimeout)
raftServer, err := raft.NewServer(testName, path, raftTransporter, store, ps, "")
if err != nil {
panic(err)
}
raftServer.SetElectionTimeout(testElectionTimeout)
raftServer.SetHeartbeatTimeout(testHeartbeatTimeout)
ps.SetRaftServer(raftServer)
s := server.New(testName, "http://"+testClientURL, ps, registry, store, nil)
sListener, err := server.NewListener(testClientURL)
if err != nil {
panic(err)
}
ps.SetServer(s)
// Start up peer server.
c := make(chan bool)
go func() {
c <- true
ps.ListenAndServe(false, []string{})
ps.Start(false, []string{})
http.Serve(psListener, ps.HTTPHandler())
}()
<-c
// Start up etcd server.
go func() {
c <- true
s.ListenAndServe()
http.Serve(sListener, s.HTTPHandler())
}()
<-c
@ -53,6 +88,7 @@ func RunServer(f func(*server.Server)) {
f(s)
// Clean up servers.
ps.Close()
s.Close()
ps.Stop()
psListener.Close()
sListener.Close()
}