This commit is contained in:
Xiang Li 2013-09-12 13:17:19 -04:00
parent 86e03d2298
commit 8d245b546f
6 changed files with 82 additions and 49 deletions

View File

@ -170,8 +170,9 @@ func (c *JoinCommand) Apply(raftServer *raft.Server) (interface{}, error) {
value := fmt.Sprintf("raft=%s&etcd=%s&raftVersion=%s", c.RaftURL, c.EtcdURL, c.RaftVersion) value := fmt.Sprintf("raft=%s&etcd=%s&raftVersion=%s", c.RaftURL, c.EtcdURL, c.RaftVersion)
etcdStore.Set(key, value, time.Unix(0, 0), raftServer.CommitIndex()) etcdStore.Set(key, value, time.Unix(0, 0), raftServer.CommitIndex())
// add peer stats
if c.Name != r.Name() { if c.Name != r.Name() {
r.peersStats[c.Name] = &raftPeerStats{MinLatency: 1 << 63} r.peersStats.Peers[c.Name] = &raftPeerStats{MinLatency: 1 << 63}
} }
return b, err return b, err
@ -198,7 +199,9 @@ func (c *RemoveCommand) Apply(raftServer *raft.Server) (interface{}, error) {
key := path.Join("_etcd/machines", c.Name) key := path.Join("_etcd/machines", c.Name)
_, err := etcdStore.Delete(key, raftServer.CommitIndex()) _, err := etcdStore.Delete(key, raftServer.CommitIndex())
delete(r.peersStats, c.Name)
// delete from stats
delete(r.peersStats.Peers, c.Name)
if err != nil { if err != nil {
return []byte{0}, err return []byte{0}, err

View File

@ -22,7 +22,7 @@ func NewEtcdMuxer() *http.ServeMux {
etcdMux.Handle("/"+version+"/watch/", errorHandler(WatchHttpHandler)) etcdMux.Handle("/"+version+"/watch/", errorHandler(WatchHttpHandler))
etcdMux.Handle("/"+version+"/leader", errorHandler(LeaderHttpHandler)) etcdMux.Handle("/"+version+"/leader", errorHandler(LeaderHttpHandler))
etcdMux.Handle("/"+version+"/machines", errorHandler(MachinesHttpHandler)) etcdMux.Handle("/"+version+"/machines", errorHandler(MachinesHttpHandler))
etcdMux.Handle("/"+version+"/stats", errorHandler(StatsHttpHandler)) etcdMux.Handle("/"+version+"/stats/", errorHandler(StatsHttpHandler))
etcdMux.Handle("/version", errorHandler(VersionHttpHandler)) etcdMux.Handle("/version", errorHandler(VersionHttpHandler))
etcdMux.HandleFunc("/test/", TestHttpHandler) etcdMux.HandleFunc("/test/", TestHttpHandler)
return etcdMux return etcdMux
@ -167,22 +167,8 @@ func dispatch(c Command, w http.ResponseWriter, req *http.Request, etcd bool) er
return etcdErr.NewError(300, "") return etcdErr.NewError(300, "")
} }
// tell the client where is the leader redirect(leader, etcd, w, req)
path := req.URL.Path
var url string
if etcd {
etcdAddr, _ := nameToEtcdURL(leader)
url = etcdAddr + path
} else {
raftAddr, _ := nameToRaftURL(leader)
url = raftAddr + path
}
debugf("Redirect to %s", url)
http.Redirect(w, req, url, http.StatusTemporaryRedirect)
return nil return nil
} }
return etcdErr.NewError(300, "") return etcdErr.NewError(300, "")
@ -227,9 +213,28 @@ func VersionHttpHandler(w http.ResponseWriter, req *http.Request) error {
// Handler to return the basic stats of etcd // Handler to return the basic stats of etcd
func StatsHttpHandler(w http.ResponseWriter, req *http.Request) error { func StatsHttpHandler(w http.ResponseWriter, req *http.Request) error {
option := req.URL.Path[len("/v1/stats/"):]
switch option {
case "self":
w.WriteHeader(http.StatusOK)
w.Write(r.Stats())
case "leader":
if r.State() == raft.Leader {
w.Write(r.PeerStats())
} else {
leader := r.Leader()
// current no leader
if leader == "" {
return etcdErr.NewError(300, "")
}
redirect(leader, true, w, req)
}
case "store":
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
w.Write(etcdStore.Stats()) w.Write(etcdStore.Stats())
w.Write(r.Stats()) }
return nil return nil
} }

View File

@ -24,7 +24,7 @@ type raftServer struct {
listenHost string listenHost string
tlsConf *TLSConfig tlsConf *TLSConfig
tlsInfo *TLSInfo tlsInfo *TLSInfo
peersStats map[string]*raftPeerStats peersStats *raftPeersStats
serverStats *raftServerStats serverStats *raftServerStats
} }
@ -48,7 +48,10 @@ func newRaftServer(name string, url string, listenHost string, tlsConf *TLSConfi
listenHost: listenHost, listenHost: listenHost,
tlsConf: tlsConf, tlsConf: tlsConf,
tlsInfo: tlsInfo, tlsInfo: tlsInfo,
peersStats: make(map[string]*raftPeerStats), peersStats: &raftPeersStats{
Leader: name,
Peers: make(map[string]*raftPeerStats),
},
serverStats: &raftServerStats{ serverStats: &raftServerStats{
StartTime: time.Now(), StartTime: time.Now(),
sendRateQueue: &statsQueue{ sendRateQueue: &statsQueue{
@ -63,7 +66,6 @@ func newRaftServer(name string, url string, listenHost string, tlsConf *TLSConfi
// Start the raft server // Start the raft server
func (r *raftServer) ListenAndServe() { func (r *raftServer) ListenAndServe() {
// Setup commands. // Setup commands.
registerCommands() registerCommands()
@ -282,7 +284,7 @@ func joinByMachine(s *raft.Server, machine string, scheme string) error {
} }
func (r *raftServer) Stats() []byte { func (r *raftServer) Stats() []byte {
r.serverStats.LeaderUptime = time.Now().Sub(r.serverStats.leaderStartTime).String() r.serverStats.LeaderInfo.Uptime = time.Now().Sub(r.serverStats.LeaderInfo.startTime).String()
queue := r.serverStats.sendRateQueue queue := r.serverStats.sendRateQueue
@ -292,20 +294,17 @@ func (r *raftServer) Stats() []byte {
r.serverStats.RecvingPkgRate, r.serverStats.RecvingBandwidthRate = queue.Rate() r.serverStats.RecvingPkgRate, r.serverStats.RecvingBandwidthRate = queue.Rate()
sBytes, err := json.Marshal(r.serverStats) b, _ := json.Marshal(r.serverStats)
if err != nil {
warn(err)
}
if r.State() == raft.Leader {
pBytes, _ := json.Marshal(r.peersStats)
b := append(sBytes, pBytes...)
return b return b
} }
return sBytes func (r *raftServer) PeerStats() []byte {
if r.State() == raft.Leader {
b, _ := json.Marshal(r.peersStats)
return b
}
return nil
} }
// Register commands to raft server // Register commands to raft server

View File

@ -35,8 +35,12 @@ func (ps *packageStats) Time() time.Time {
type raftServerStats struct { type raftServerStats struct {
State string `json:"state"` State string `json:"state"`
StartTime time.Time `json:"startTime"` StartTime time.Time `json:"startTime"`
Leader string `json:"leader"`
LeaderUptime string `json:"leaderUptime"` LeaderInfo struct {
Name string `json:"leader"`
Uptime string `json:"uptime"`
startTime time.Time
} `json:"leaderInfo"`
RecvAppendRequestCnt uint64 `json:"recvAppendRequestCnt,"` RecvAppendRequestCnt uint64 `json:"recvAppendRequestCnt,"`
RecvingPkgRate float64 `json:"recvPkgRate,omitempty"` RecvingPkgRate float64 `json:"recvPkgRate,omitempty"`
@ -46,16 +50,15 @@ type raftServerStats struct {
SendingPkgRate float64 `json:"sendPkgRate,omitempty"` SendingPkgRate float64 `json:"sendPkgRate,omitempty"`
SendingBandwidthRate float64 `json:"sendBandwidthRate,omitempty"` SendingBandwidthRate float64 `json:"sendBandwidthRate,omitempty"`
leaderStartTime time.Time
sendRateQueue *statsQueue sendRateQueue *statsQueue
recvRateQueue *statsQueue recvRateQueue *statsQueue
} }
func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) { func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) {
ss.State = raft.Follower ss.State = raft.Follower
if leaderName != ss.Leader { if leaderName != ss.LeaderInfo.Name {
ss.Leader = leaderName ss.LeaderInfo.Name = leaderName
ss.leaderStartTime = time.Now() ss.LeaderInfo.startTime = time.Now()
} }
ss.recvRateQueue.Insert(NewPackageStats(time.Now(), pkgSize)) ss.recvRateQueue.Insert(NewPackageStats(time.Now(), pkgSize))
@ -64,17 +67,23 @@ func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) {
func (ss *raftServerStats) SendAppendReq(pkgSize int) { func (ss *raftServerStats) SendAppendReq(pkgSize int) {
now := time.Now() now := time.Now()
if ss.State != raft.Leader { if ss.State != raft.Leader {
ss.State = raft.Leader ss.State = raft.Leader
ss.Leader = r.Name() ss.LeaderInfo.Name = r.Name()
ss.leaderStartTime = now ss.LeaderInfo.startTime = now
} }
ss.sendRateQueue.Insert(NewPackageStats(time.Now(), pkgSize)) ss.sendRateQueue.Insert(NewPackageStats(now, pkgSize))
ss.SendAppendRequestCnt++ ss.SendAppendRequestCnt++
} }
type raftPeersStats struct {
Leader string `json:"leader"`
Peers map[string]*raftPeerStats `json:"peers"`
}
type raftPeerStats struct { type raftPeerStats struct {
Latency float64 `json:"latency"` Latency float64 `json:"latency"`
AvgLatency float64 `json:"averageLatency"` AvgLatency float64 `json:"averageLatency"`

View File

@ -66,7 +66,7 @@ func (t *transporter) SendAppendEntriesRequest(server *raft.Server, peer *raft.P
debugf("Send LogEntries to %s ", u) debugf("Send LogEntries to %s ", u)
thisPeerStats, ok := r.peersStats[peer.Name] thisPeerStats, ok := r.peersStats.Peers[peer.Name]
start := time.Now() start := time.Now()
@ -85,7 +85,7 @@ func (t *transporter) SendAppendEntriesRequest(server *raft.Server, peer *raft.P
} }
} }
r.peersStats[peer.Name] = thisPeerStats r.peersStats.Peers[peer.Name] = thisPeerStats
if resp != nil { if resp != nil {
defer resp.Body.Close() defer resp.Body.Close()

17
util.go
View File

@ -128,6 +128,23 @@ func sanitizeListenHost(listen string, advertised string) string {
return net.JoinHostPort(listen, aport) return net.JoinHostPort(listen, aport)
} }
func redirect(node string, etcd bool, w http.ResponseWriter, req *http.Request) {
var url string
path := req.URL.Path
if etcd {
etcdAddr, _ := nameToEtcdURL(node)
url = etcdAddr + path
} else {
raftAddr, _ := nameToRaftURL(node)
url = raftAddr + path
}
debugf("Redirect to %s", url)
http.Redirect(w, req, url, http.StatusTemporaryRedirect)
}
func check(err error) { func check(err error) {
if err != nil { if err != nil {
fatal(err) fatal(err)