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

Merge upstream

release-0.4
David Fisher 2013-09-26 13:05:09 -07:00
parent 0b37c808dd 21f6b50607
commit 9825976e06
37 changed files with 372 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -1,4 +1,5 @@
src/
pkg/
./etcd
/etcd
release_version.go
/machine*

2
build
View File

@ -1,4 +1,4 @@
#!/bin/bash
#!/bin/sh
ETCD_PACKAGE=github.com/coreos/etcd
export GOPATH="${PWD}"

8
command.go
View File

@ -170,8 +170,10 @@ func (c *JoinCommand) Apply(raftServer *raft.Server) (interface{}, error) {
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())
// add peer stats
if c.Name != r.Name() {
r.peersStats[c.Name] = &raftPeerStats{MinLatency: 1 << 63}
r.peersStats.Peers[c.Name] = &raftPeerStats{}
r.peersStats.Peers[c.Name].Latency.Minimum = 1 << 63
}
return b, err
@ -198,7 +200,9 @@ func (c *RemoveCommand) Apply(raftServer *raft.Server) (interface{}, error) {
key := path.Join("_etcd/machines", c.Name)
_, err := etcdStore.Delete(key, raftServer.CommitIndex())
delete(r.peersStats, c.Name)
// delete from stats
delete(r.peersStats.Peers, c.Name)
if err != nil {
return []byte{0}, err

43
etcd_handlers.go
View File

@ -22,7 +22,7 @@ func NewEtcdMuxer() *http.ServeMux {
etcdMux.Handle("/"+version+"/watch/", errorHandler(WatchHttpHandler))
etcdMux.Handle("/"+version+"/leader", errorHandler(LeaderHttpHandler))
etcdMux.Handle("/"+version+"/machines", errorHandler(MachinesHttpHandler))
etcdMux.Handle("/"+version+"/stats", errorHandler(StatsHttpHandler))
etcdMux.Handle("/"+version+"/stats/", errorHandler(StatsHttpHandler))
etcdMux.Handle("/version", errorHandler(VersionHttpHandler))
etcdMux.HandleFunc("/test/", TestHttpHandler)
return etcdMux
@ -167,22 +167,8 @@ func dispatch(c Command, w http.ResponseWriter, req *http.Request, etcd bool) er
return etcdErr.NewError(300, "")
}
// tell the client where is the leader
path := req.URL.Path
redirect(leader, etcd, w, req)
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 etcdErr.NewError(300, "")
@ -227,9 +213,28 @@ func VersionHttpHandler(w http.ResponseWriter, req *http.Request) error {
// Handler to return the basic stats of etcd
func StatsHttpHandler(w http.ResponseWriter, req *http.Request) error {
w.WriteHeader(http.StatusOK)
w.Write(etcdStore.Stats())
w.Write(r.Stats())
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.Write(etcdStore.Stats())
}
return nil
}

27
raft_server.go
View File

@ -24,7 +24,7 @@ type raftServer struct {
listenHost string
tlsConf *TLSConfig
tlsInfo *TLSInfo
peersStats map[string]*raftPeerStats
peersStats *raftPeersStats
serverStats *raftServerStats
}
@ -36,7 +36,7 @@ func newRaftServer(name string, url string, listenHost string, tlsConf *TLSConfi
raftTransporter := newTransporter(tlsConf.Scheme, tlsConf.Client, ElectionTimeout)
// Create raft server
server, err := raft.NewServer(name, dirPath, raftTransporter, etcdStore, nil)
server, err := raft.NewServer(name, dirPath, raftTransporter, etcdStore, nil, "")
check(err)
@ -48,7 +48,10 @@ func newRaftServer(name string, url string, listenHost string, tlsConf *TLSConfi
listenHost: listenHost,
tlsConf: tlsConf,
tlsInfo: tlsInfo,
peersStats: make(map[string]*raftPeerStats),
peersStats: &raftPeersStats{
Leader: name,
Peers: make(map[string]*raftPeerStats),
},
serverStats: &raftServerStats{
StartTime: time.Now(),
sendRateQueue: &statsQueue{
@ -63,7 +66,6 @@ func newRaftServer(name string, url string, listenHost string, tlsConf *TLSConfi
// Start the raft server
func (r *raftServer) ListenAndServe() {
// Setup commands.
registerCommands()
@ -282,7 +284,7 @@ func joinByMachine(s *raft.Server, machine string, scheme string) error {
}
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
@ -292,20 +294,17 @@ func (r *raftServer) Stats() []byte {
r.serverStats.RecvingPkgRate, r.serverStats.RecvingBandwidthRate = queue.Rate()
sBytes, err := json.Marshal(r.serverStats)
b, _ := json.Marshal(r.serverStats)
if err != nil {
warn(err)
}
return b
}
func (r *raftServer) PeerStats() []byte {
if r.State() == raft.Leader {
pBytes, _ := json.Marshal(r.peersStats)
b := append(sBytes, pBytes...)
b, _ := json.Marshal(r.peersStats)
return b
}
return sBytes
return nil
}
// Register commands to raft server

80
raft_stats.go
View File

@ -33,10 +33,14 @@ func (ps *packageStats) Time() time.Time {
}
type raftServerStats struct {
State string `json:"state"`
StartTime time.Time `json:"startTime"`
Leader string `json:"leader"`
LeaderUptime string `json:"leaderUptime"`
State string `json:"state"`
StartTime time.Time `json:"startTime"`
LeaderInfo struct {
Name string `json:"leader"`
Uptime string `json:"uptime"`
startTime time.Time
} `json:"leaderInfo"`
RecvAppendRequestCnt uint64 `json:"recvAppendRequestCnt,"`
RecvingPkgRate float64 `json:"recvPkgRate,omitempty"`
@ -46,16 +50,15 @@ type raftServerStats struct {
SendingPkgRate float64 `json:"sendPkgRate,omitempty"`
SendingBandwidthRate float64 `json:"sendBandwidthRate,omitempty"`
leaderStartTime time.Time
sendRateQueue *statsQueue
recvRateQueue *statsQueue
sendRateQueue *statsQueue
recvRateQueue *statsQueue
}
func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) {
ss.State = raft.Follower
if leaderName != ss.Leader {
ss.Leader = leaderName
ss.leaderStartTime = time.Now()
if leaderName != ss.LeaderInfo.Name {
ss.LeaderInfo.Name = leaderName
ss.LeaderInfo.startTime = time.Now()
}
ss.recvRateQueue.Insert(NewPackageStats(time.Now(), pkgSize))
@ -64,55 +67,66 @@ func (ss *raftServerStats) RecvAppendReq(leaderName string, pkgSize int) {
func (ss *raftServerStats) SendAppendReq(pkgSize int) {
now := time.Now()
if ss.State != raft.Leader {
ss.State = raft.Leader
ss.Leader = r.Name()
ss.leaderStartTime = now
ss.LeaderInfo.Name = r.Name()
ss.LeaderInfo.startTime = now
}
ss.sendRateQueue.Insert(NewPackageStats(time.Now(), pkgSize))
ss.sendRateQueue.Insert(NewPackageStats(now, pkgSize))
ss.SendAppendRequestCnt++
}
type raftPeersStats struct {
Leader string `json:"leader"`
Peers map[string]*raftPeerStats `json:"peers"`
}
type raftPeerStats struct {
Latency float64 `json:"latency"`
AvgLatency float64 `json:"averageLatency"`
avgLatencySquare float64
SdvLatency float64 `json:"sdvLatency"`
MinLatency float64 `json:"minLatency"`
MaxLatency float64 `json:"maxLatency"`
FailCnt uint64 `json:"failsCount"`
SuccCnt uint64 `json:"successCount"`
Latency struct {
Current float64 `json:"current"`
Average float64 `json:"average"`
averageSquare float64
StandardDeviation float64 `json:"standardDeviation"`
Minimum float64 `json:"minimum"`
Maximum float64 `json:"maximum"`
} `json:"latency"`
Counts struct {
Fail uint64 `json:"fail"`
Success uint64 `json:"success"`
} `json:"counts"`
}
// Succ function update the raftPeerStats with a successful send
func (ps *raftPeerStats) Succ(d time.Duration) {
total := float64(ps.SuccCnt) * ps.AvgLatency
totalSquare := float64(ps.SuccCnt) * ps.avgLatencySquare
total := float64(ps.Counts.Success) * ps.Latency.Average
totalSquare := float64(ps.Counts.Success) * ps.Latency.averageSquare
ps.SuccCnt++
ps.Counts.Success++
ps.Latency = float64(d) / (1000000.0)
ps.Latency.Current = float64(d) / (1000000.0)
if ps.Latency > ps.MaxLatency {
ps.MaxLatency = ps.Latency
if ps.Latency.Current > ps.Latency.Maximum {
ps.Latency.Maximum = ps.Latency.Current
}
if ps.Latency < ps.MinLatency {
ps.MinLatency = ps.Latency
if ps.Latency.Current < ps.Latency.Minimum {
ps.Latency.Minimum = ps.Latency.Current
}
ps.AvgLatency = (total + ps.Latency) / float64(ps.SuccCnt)
ps.avgLatencySquare = (totalSquare + ps.Latency*ps.Latency) / float64(ps.SuccCnt)
ps.Latency.Average = (total + ps.Latency.Current) / float64(ps.Counts.Success)
ps.Latency.averageSquare = (totalSquare + ps.Latency.Current * ps.Latency.Current) / float64(ps.Counts.Success)
// sdv = sqrt(avg(x^2) - avg(x)^2)
ps.SdvLatency = math.Sqrt(ps.avgLatencySquare - ps.AvgLatency*ps.AvgLatency)
ps.Latency.StandardDeviation = math.Sqrt(ps.Latency.averageSquare - ps.Latency.Average*ps.Latency.Average)
}
// Fail function update the raftPeerStats with a unsuccessful send
func (ps *raftPeerStats) Fail() {
ps.FailCnt++
ps.Counts.Fail++
}
type statsQueue struct {

19
scripts/test-cluster Executable file
View File

@ -0,0 +1,19 @@
#!/bin/bash
SESSION=etcd-cluster
tmux new-session -d -s $SESSION
# Setup a window for tailing log files
tmux new-window -t $SESSION:1 -n 'machines'
tmux split-window -h
tmux select-pane -t 0
tmux send-keys "./etcd -s 127.0.0.1:7001 -c 127.0.0.1:4001 -d machine1 -n machine1" C-m
for i in 2 3; do
tmux select-pane -t 0
tmux split-window -v
tmux send-keys "./etcd -cors='*' -s 127.0.0.1:700${i} -c 127.0.0.1:400${i} -C 127.0.0.1:7001 -d machine${i} -n machine${i}" C-m
done
# Attach to session
tmux attach-session -t $SESSION

70
third_party/code.google.com/p/goprotobuf/proto/all_test.go vendored
View File

@ -431,7 +431,7 @@ func TestRequiredBit(t *testing.T) {
err := o.Marshal(pb)
if err == nil {
t.Error("did not catch missing required fields")
} else if strings.Index(err.Error(), "GoTest") < 0 {
} else if strings.Index(err.Error(), "Kind") < 0 {
t.Error("wrong error type:", err)
}
}
@ -1205,7 +1205,7 @@ func TestRequiredFieldEnforcement(t *testing.T) {
_, err := Marshal(pb)
if err == nil {
t.Error("marshal: expected error, got nil")
} else if strings.Index(err.Error(), "GoTestField") < 0 {
} else if strings.Index(err.Error(), "Label") < 0 {
t.Errorf("marshal: bad error type: %v", err)
}
@ -1216,7 +1216,7 @@ func TestRequiredFieldEnforcement(t *testing.T) {
err = Unmarshal(buf, pb)
if err == nil {
t.Error("unmarshal: expected error, got nil")
} else if strings.Index(err.Error(), "GoTestField") < 0 {
} else if strings.Index(err.Error(), "{Unknown}") < 0 {
t.Errorf("unmarshal: bad error type: %v", err)
}
}
@ -1670,6 +1670,70 @@ func TestEncodingSizes(t *testing.T) {
}
}
func TestErrRequiredNotSet(t *testing.T) {
pb := initGoTest(false)
pb.RequiredField.Label = nil
pb.F_Int32Required = nil
pb.F_Int64Required = nil
expected := "0807" + // field 1, encoding 0, value 7
"2206" + "120474797065" + // field 4, encoding 2 (GoTestField)
"5001" + // field 10, encoding 0, value 1
"6d20000000" + // field 13, encoding 5, value 0x20
"714000000000000000" + // field 14, encoding 1, value 0x40
"78a019" + // field 15, encoding 0, value 0xca0 = 3232
"8001c032" + // field 16, encoding 0, value 0x1940 = 6464
"8d0100004a45" + // field 17, encoding 5, value 3232.0
"9101000000000040b940" + // field 18, encoding 1, value 6464.0
"9a0106" + "737472696e67" + // field 19, encoding 2, string "string"
"b304" + // field 70, encoding 3, start group
"ba0408" + "7265717569726564" + // field 71, encoding 2, string "required"
"b404" + // field 70, encoding 4, end group
"aa0605" + "6279746573" + // field 101, encoding 2, string "bytes"
"b0063f" + // field 102, encoding 0, 0x3f zigzag32
"b8067f" // field 103, encoding 0, 0x7f zigzag64
o := old()
bytes, err := Marshal(pb)
if _, ok := err.(*ErrRequiredNotSet); !ok {
fmt.Printf("marshal-1 err = %v, want *ErrRequiredNotSet", err)
o.DebugPrint("", bytes)
t.Fatalf("expected = %s", expected)
}
if strings.Index(err.Error(), "RequiredField.Label") < 0 {
t.Errorf("marshal-1 wrong err msg: %v", err)
}
if !equal(bytes, expected, t) {
o.DebugPrint("neq 1", bytes)
t.Fatalf("expected = %s", expected)
}
// Now test Unmarshal by recreating the original buffer.
pbd := new(GoTest)
err = Unmarshal(bytes, pbd)
if _, ok := err.(*ErrRequiredNotSet); !ok {
t.Fatalf("unmarshal err = %v, want *ErrRequiredNotSet", err)
o.DebugPrint("", bytes)
t.Fatalf("string = %s", expected)
}
if strings.Index(err.Error(), "RequiredField.{Unknown}") < 0 {
t.Errorf("unmarshal wrong err msg: %v", err)
}
bytes, err = Marshal(pbd)
if _, ok := err.(*ErrRequiredNotSet); !ok {
t.Errorf("marshal-2 err = %v, want *ErrRequiredNotSet", err)
o.DebugPrint("", bytes)
t.Fatalf("string = %s", expected)
}
if strings.Index(err.Error(), "RequiredField.Label") < 0 {
t.Errorf("marshal-2 wrong err msg: %v", err)
}
if !equal(bytes, expected, t) {
o.DebugPrint("neq 2", bytes)
t.Fatalf("string = %s", expected)
}
}
func fuzzUnmarshal(t *testing.T, data []byte) {
defer func() {
if e := recover(); e != nil {

16
third_party/code.google.com/p/goprotobuf/proto/decode.go vendored
View File

@ -46,7 +46,7 @@ import (
// ErrWrongType occurs when the wire encoding for the field disagrees with
// that specified in the type being decoded. This is usually caused by attempting
// to convert an encoded protocol buffer into a struct of the wrong type.
var ErrWrongType = errors.New("field/encoding mismatch: wrong type for field")
var ErrWrongType = errors.New("proto: field/encoding mismatch: wrong type for field")
// errOverflow is returned when an integer is too large to be represented.
var errOverflow = errors.New("proto: integer overflow")
@ -353,6 +353,7 @@ func (p *Buffer) Unmarshal(pb Message) error {
// unmarshalType does the work of unmarshaling a structure.
func (o *Buffer) unmarshalType(st reflect.Type, prop *StructProperties, is_group bool, base structPointer) error {
var state errorState
required, reqFields := prop.reqCount, uint64(0)
var err error
@ -406,7 +407,10 @@ func (o *Buffer) unmarshalType(st reflect.Type, prop *StructProperties, is_group
continue
}
}
err = dec(o, p, base)
decErr := dec(o, p, base)
if decErr != nil && !state.shouldContinue(decErr, p) {
err = decErr
}
if err == nil && p.Required {
// Successfully decoded a required field.
if tag <= 64 {
@ -430,8 +434,14 @@ func (o *Buffer) unmarshalType(st reflect.Type, prop *StructProperties, is_group
if is_group {
return io.ErrUnexpectedEOF
}
if state.err != nil {
return state.err
}
if required > 0 {
return &ErrRequiredNotSet{st}
// Not enough information to determine the exact field. If we use extra
// CPU, we could determine the field only if the missing required field
// has a tag <= 64 and we check reqFields.
return &ErrRequiredNotSet{"{Unknown}"}
}
}
return err

78
third_party/code.google.com/p/goprotobuf/proto/encode.go vendored
View File

@ -37,6 +37,7 @@ package proto
import (
"errors"
"fmt"
"reflect"
"sort"
)
@ -46,12 +47,16 @@ import (
// all been initialized. It is also the error returned if Unmarshal is
// called with an encoded protocol buffer that does not include all the
// required fields.
//
// When printed, ErrRequiredNotSet reports the first unset required field in a
// message. If the field cannot be precisely determined, it is reported as
// "{Unknown}".
type ErrRequiredNotSet struct {
t reflect.Type
field string
}
func (e *ErrRequiredNotSet) Error() string {
return "proto: required fields not set in " + e.t.String()
return fmt.Sprintf("proto: required field %q not set", e.field)
}
var (
@ -175,7 +180,8 @@ func Marshal(pb Message) ([]byte, error) {
}
p := NewBuffer(nil)
err := p.Marshal(pb)
if err != nil {
var state errorState
if err != nil && !state.shouldContinue(err, nil) {
return nil, err
}
return p.buf, err
@ -274,6 +280,7 @@ func isNil(v reflect.Value) bool {
// Encode a message struct.
func (o *Buffer) enc_struct_message(p *Properties, base structPointer) error {
var state errorState
structp := structPointer_GetStructPointer(base, p.field)
if structPointer_IsNil(structp) {
return ErrNil
@ -283,7 +290,7 @@ func (o *Buffer) enc_struct_message(p *Properties, base structPointer) error {
if p.isMarshaler {
m := structPointer_Interface(structp, p.stype).(Marshaler)
data, err := m.Marshal()
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
return err
}
o.buf = append(o.buf, p.tagcode...)
@ -300,18 +307,19 @@ func (o *Buffer) enc_struct_message(p *Properties, base structPointer) error {
nbuf := o.buf
o.buf = obuf
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
o.buffree(nbuf)
return err
}
o.buf = append(o.buf, p.tagcode...)
o.EncodeRawBytes(nbuf)
o.buffree(nbuf)
return nil
return state.err
}
// Encode a group struct.
func (o *Buffer) enc_struct_group(p *Properties, base structPointer) error {
var state errorState
b := structPointer_GetStructPointer(base, p.field)
if structPointer_IsNil(b) {
return ErrNil
@ -319,11 +327,11 @@ func (o *Buffer) enc_struct_group(p *Properties, base structPointer) error {
o.EncodeVarint(uint64((p.Tag << 3) | WireStartGroup))
err := o.enc_struct(p.stype, p.sprop, b)
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
return err
}
o.EncodeVarint(uint64((p.Tag << 3) | WireEndGroup))
return nil
return state.err
}
// Encode a slice of bools ([]bool).
@ -470,6 +478,7 @@ func (o *Buffer) enc_slice_string(p *Properties, base structPointer) error {
// Encode a slice of message structs ([]*struct).
func (o *Buffer) enc_slice_struct_message(p *Properties, base structPointer) error {
var state errorState
s := structPointer_StructPointerSlice(base, p.field)
l := s.Len()
@ -483,7 +492,7 @@ func (o *Buffer) enc_slice_struct_message(p *Properties, base structPointer) err
if p.isMarshaler {
m := structPointer_Interface(structp, p.stype).(Marshaler)
data, err := m.Marshal()
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
return err
}
o.buf = append(o.buf, p.tagcode...)
@ -498,7 +507,7 @@ func (o *Buffer) enc_slice_struct_message(p *Properties, base structPointer) err
nbuf := o.buf
o.buf = obuf
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
o.buffree(nbuf)
if err == ErrNil {
return ErrRepeatedHasNil
@ -510,11 +519,12 @@ func (o *Buffer) enc_slice_struct_message(p *Properties, base structPointer) err
o.buffree(nbuf)
}
return nil
return state.err
}
// Encode a slice of group structs ([]*struct).
func (o *Buffer) enc_slice_struct_group(p *Properties, base structPointer) error {
var state errorState
s := structPointer_StructPointerSlice(base, p.field)
l := s.Len()
@ -528,7 +538,7 @@ func (o *Buffer) enc_slice_struct_group(p *Properties, base structPointer) error
err := o.enc_struct(p.stype, p.sprop, b)
if err != nil {
if err != nil && !state.shouldContinue(err, nil) {
if err == ErrNil {
return ErrRepeatedHasNil
}
@ -537,7 +547,7 @@ func (o *Buffer) enc_slice_struct_group(p *Properties, base structPointer) error
o.EncodeVarint(uint64((p.Tag << 3) | WireEndGroup))
}
return nil
return state.err
}
// Encode an extension map.
@ -569,7 +579,7 @@ func (o *Buffer) enc_map(p *Properties, base structPointer) error {
// Encode a struct.
func (o *Buffer) enc_struct(t reflect.Type, prop *StructProperties, base structPointer) error {
required := prop.reqCount
var state errorState
// Encode fields in tag order so that decoders may use optimizations
// that depend on the ordering.
// http://code.google.com/apis/protocolbuffers/docs/encoding.html#order
@ -577,19 +587,15 @@ func (o *Buffer) enc_struct(t reflect.Type, prop *StructProperties, base structP
p := prop.Prop[i]
if p.enc != nil {
err := p.enc(o, p, base)
if err != nil {
if err != nil && !state.shouldContinue(err, p) {
if err != ErrNil {
return err
} else if p.Required && state.err == nil {
state.err = &ErrRequiredNotSet{p.Name}
}
} else if p.Required {
required--
}
}
}
// See if we encoded all required fields.
if required > 0 {
return &ErrRequiredNotSet{t}
}
// Add unrecognized fields at the end.
if prop.unrecField.IsValid() {
@ -599,5 +605,33 @@ func (o *Buffer) enc_struct(t reflect.Type, prop *StructProperties, base structP
}
}
return nil
return state.err
}
// errorState maintains the first error that occurs and updates that error
// with additional context.
type errorState struct {
err error
}
// shouldContinue reports whether encoding should continue upon encountering the
// given error. If the error is ErrRequiredNotSet, shouldContinue returns true
// and, if this is the first appearance of that error, remembers it for future
// reporting.
//
// If prop is not nil, it may update any error with additional context about the
// field with the error.
func (s *errorState) shouldContinue(err error, prop *Properties) bool {
// Ignore unset required fields.
reqNotSet, ok := err.(*ErrRequiredNotSet)
if !ok {
return false
}
if s.err == nil {
if prop != nil {
err = &ErrRequiredNotSet{prop.Name + "." + reqNotSet.field}
}
s.err = err
}
return true
}

8
third_party/code.google.com/p/goprotobuf/proto/extensions.go vendored
View File

@ -109,11 +109,11 @@ func isExtensionField(pb extendableProto, field int32) bool {
func checkExtensionTypes(pb extendableProto, extension *ExtensionDesc) error {
// Check the extended type.
if a, b := reflect.TypeOf(pb), reflect.TypeOf(extension.ExtendedType); a != b {
return errors.New("bad extended type; " + b.String() + " does not extend " + a.String())
return errors.New("proto: bad extended type; " + b.String() + " does not extend " + a.String())
}
// Check the range.
if !isExtensionField(pb, extension.Field) {
return errors.New("bad extension number; not in declared ranges")
return errors.New("proto: bad extension number; not in declared ranges")
}
return nil
}
@ -272,7 +272,7 @@ func decodeExtension(b []byte, extension *ExtensionDesc) (interface{}, error) {
func GetExtensions(pb Message, es []*ExtensionDesc) (extensions []interface{}, err error) {
epb, ok := pb.(extendableProto)
if !ok {
err = errors.New("not an extendable proto")
err = errors.New("proto: not an extendable proto")
return
}
extensions = make([]interface{}, len(es))
@ -292,7 +292,7 @@ func SetExtension(pb extendableProto, extension *ExtensionDesc, value interface{
}
typ := reflect.TypeOf(extension.ExtensionType)
if typ != reflect.TypeOf(value) {
return errors.New("bad extension value type")
return errors.New("proto: bad extension value type")
}
pb.ExtensionMap()[extension.Field] = Extension{desc: extension, value: value}

6
third_party/code.google.com/p/goprotobuf/proto/testdata/test.pb.go vendored
View File

@ -244,7 +244,7 @@ func (m *GoEnum) GetFoo() FOO {
if m != nil && m.Foo != nil {
return *m.Foo
}
return 0
return FOO_FOO1
}
type GoTestField struct {
@ -378,7 +378,7 @@ func (m *GoTest) GetKind() GoTest_KIND {
if m != nil && m.Kind != nil {
return *m.Kind
}
return 0
return GoTest_VOID
}
func (m *GoTest) GetTable() string {
@ -1289,7 +1289,7 @@ func (m *MyMessage) GetBikeshed() MyMessage_Color {
if m != nil && m.Bikeshed != nil {
return *m.Bikeshed
}
return 0
return MyMessage_RED
}
func (m *MyMessage) GetSomegroup() *MyMessage_SomeGroup {

4
third_party/code.google.com/p/goprotobuf/proto/text_parser.go vendored
View File

@ -193,8 +193,8 @@ func (p *textParser) advance() {
}
var (
errBadUTF8 = errors.New("bad UTF-8")
errBadHex = errors.New("bad hexadecimal")
errBadUTF8 = errors.New("proto: bad UTF-8")
errBadHex = errors.New("proto: bad hexadecimal")
)
func unquoteC(s string, quote rune) (string, error) {

2
third_party/code.google.com/p/goprotobuf/proto/text_test.go vendored
View File

@ -303,7 +303,7 @@ type limitedWriter struct {
limit int
}
var outOfSpace = errors.New("insufficient space")
var outOfSpace = errors.New("proto: insufficient space")
func (w *limitedWriter) Write(p []byte) (n int, err error) {
var avail = w.limit - w.b.Len()

4
third_party/code.google.com/p/goprotobuf/protoc-gen-go/descriptor/descriptor.pb.go vendored
View File

@ -487,14 +487,14 @@ func (m *FieldDescriptorProto) GetLabel() FieldDescriptorProto_Label {
if m != nil && m.Label != nil {
return *m.Label
}
return 0
return FieldDescriptorProto_LABEL_OPTIONAL
}
func (m *FieldDescriptorProto) GetType() FieldDescriptorProto_Type {
if m != nil && m.Type != nil {
return *m.Type
}
return 0
return FieldDescriptorProto_TYPE_DOUBLE
}
func (m *FieldDescriptorProto) GetTypeName() string {

21
third_party/code.google.com/p/goprotobuf/protoc-gen-go/generator/generator.go vendored
View File

@ -1664,6 +1664,27 @@ func (g *Generator) generateMessage(message *Descriptor) {
g.P("return false")
case descriptor.FieldDescriptorProto_TYPE_STRING:
g.P(`return ""`)
case descriptor.FieldDescriptorProto_TYPE_ENUM:
// The default default for an enum is the first value in the enum,
// not zero.
obj := g.ObjectNamed(field.GetTypeName())
var enum *EnumDescriptor
if id, ok := obj.(*ImportedDescriptor); ok {
// The enum type has been publicly imported.
enum, _ = id.o.(*EnumDescriptor)
} else {
enum, _ = obj.(*EnumDescriptor)
}
if enum == nil {
log.Printf("don't know how to generate getter for %s", field.GetName())
continue
}
if len(enum.Value) == 0 {
g.P("return 0 // empty enum")
} else {
first := enum.Value[0].GetName()
g.P("return ", g.DefaultPackageName(obj)+enum.prefix()+first)
}
default:
g.P("return 0")
}

2
third_party/code.google.com/p/goprotobuf/protoc-gen-go/testdata/my_test/test.pb.go vendored
View File

@ -199,7 +199,7 @@ func (m *Request) GetHue() Request_Color {
if m != nil && m.Hue != nil {
return *m.Hue
}
return 0
return Request_RED
}
func (m *Request) GetHat() HatType {

2
third_party/code.google.com/p/goprotobuf/protoc-gen-go/testdata/my_test/test.pb.go.golden vendored
View File

@ -199,7 +199,7 @@ func (m *Request) GetHue() Request_Color {
if m != nil && m.Hue != nil {
return *m.Hue
}
return 0
return Request_RED
}
func (m *Request) GetHat() HatType {

2
third_party/code.google.com/p/goprotobuf/protoc-gen-go/testdata/my_test/test.proto vendored
View File

@ -58,7 +58,7 @@ message Request {
}
repeated int64 key = 1;
// optional imp.ImportedMessage imported_message = 2;
optional Color hue = 3;
optional Color hue = 3; // no default
optional HatType hat = 4 [default=FEDORA];
// optional imp.ImportedMessage.Owner owner = 6;
optional float deadline = 7 [default=inf];

4
third_party/github.com/coreos/go-raft/append_entries_request.go vendored
View File

@ -31,7 +31,7 @@ func newAppendEntriesRequest(term uint64, prevLogIndex uint64, prevLogTerm uint6
// Encodes the AppendEntriesRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *AppendEntriesRequest) encode(w io.Writer) (int, error) {
func (req *AppendEntriesRequest) Encode(w io.Writer) (int, error) {
protoEntries := make([]*protobuf.ProtoAppendEntriesRequest_ProtoLogEntry, len(req.Entries))
@ -63,7 +63,7 @@ func (req *AppendEntriesRequest) encode(w io.Writer) (int, error) {
// Decodes the AppendEntriesRequest from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *AppendEntriesRequest) decode(r io.Reader) (int, error) {
func (req *AppendEntriesRequest) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

6
third_party/github.com/coreos/go-raft/append_entries_request_test.go vendored
View File

@ -10,7 +10,7 @@ func BenchmarkAppendEntriesRequestEncoding(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
var buf bytes.Buffer
req.encode(&buf)
req.Encode(&buf)
}
b.SetBytes(int64(len(tmp)))
}
@ -19,7 +19,7 @@ func BenchmarkAppendEntriesRequestDecoding(b *testing.B) {
req, buf := createTestAppendEntriesRequest(2000)
b.ResetTimer()
for i := 0; i < b.N; i++ {
req.decode(bytes.NewReader(buf))
req.Decode(bytes.NewReader(buf))
}
b.SetBytes(int64(len(buf)))
}
@ -34,7 +34,7 @@ func createTestAppendEntriesRequest(entryCount int) (*AppendEntriesRequest, []by
req := newAppendEntriesRequest(1, 1, 1, 1, "leader", entries)
var buf bytes.Buffer
req.encode(&buf)
req.Encode(&buf)
return req, buf.Bytes()
}

4
third_party/github.com/coreos/go-raft/append_entries_response.go vendored
View File

@ -30,7 +30,7 @@ func newAppendEntriesResponse(term uint64, success bool, index uint64, commitInd
// Encodes the AppendEntriesResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (resp *AppendEntriesResponse) encode(w io.Writer) (int, error) {
func (resp *AppendEntriesResponse) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoAppendEntriesResponse{
Term: proto.Uint64(resp.Term),
Index: proto.Uint64(resp.Index),
@ -47,7 +47,7 @@ func (resp *AppendEntriesResponse) encode(w io.Writer) (int, error) {
// Decodes the AppendEntriesResponse from a buffer. Returns the number of bytes read and
// any error that occurs.
func (resp *AppendEntriesResponse) decode(r io.Reader) (int, error) {
func (resp *AppendEntriesResponse) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

6
third_party/github.com/coreos/go-raft/append_entries_response_test.go vendored
View File

@ -10,7 +10,7 @@ func BenchmarkAppendEntriesResponseEncoding(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
var buf bytes.Buffer
req.encode(&buf)
req.Encode(&buf)
}
b.SetBytes(int64(len(tmp)))
}
@ -19,7 +19,7 @@ func BenchmarkAppendEntriesResponseDecoding(b *testing.B) {
req, buf := createTestAppendEntriesResponse(2000)
b.ResetTimer()
for i := 0; i < b.N; i++ {
req.decode(bytes.NewReader(buf))
req.Decode(bytes.NewReader(buf))
}
b.SetBytes(int64(len(buf)))
}
@ -28,7 +28,7 @@ func createTestAppendEntriesResponse(entryCount int) (*AppendEntriesResponse, []
resp := newAppendEntriesResponse(1, true, 1, 1)
var buf bytes.Buffer
resp.encode(&buf)
resp.Encode(&buf)
return resp, buf.Bytes()
}

16
third_party/github.com/coreos/go-raft/http_transporter.go vendored
View File

@ -89,7 +89,7 @@ func (t *HTTPTransporter) Install(server *Server, mux HTTPMuxer) {
// Sends an AppendEntries RPC to a peer.
func (t *HTTPTransporter) SendAppendEntriesRequest(server *Server, peer *Peer, req *AppendEntriesRequest) *AppendEntriesResponse {
var b bytes.Buffer
if _, err := req.encode(&b); err != nil {
if _, err := req.Encode(&b); err != nil {
traceln("transporter.ae.encoding.error:", err)
return nil
}
@ -106,7 +106,7 @@ func (t *HTTPTransporter) SendAppendEntriesRequest(server *Server, peer *Peer, r
defer httpResp.Body.Close()
resp := &AppendEntriesResponse{}
if _, err = resp.decode(httpResp.Body); err != nil && err != io.EOF {
if _, err = resp.Decode(httpResp.Body); err != nil && err != io.EOF {
traceln("transporter.ae.decoding.error:", err)
return nil
}
@ -117,7 +117,7 @@ func (t *HTTPTransporter) SendAppendEntriesRequest(server *Server, peer *Peer, r
// Sends a RequestVote RPC to a peer.
func (t *HTTPTransporter) SendVoteRequest(server *Server, peer *Peer, req *RequestVoteRequest) *RequestVoteResponse {
var b bytes.Buffer
if _, err := req.encode(&b); err != nil {
if _, err := req.Encode(&b); err != nil {
traceln("transporter.rv.encoding.error:", err)
return nil
}
@ -134,7 +134,7 @@ func (t *HTTPTransporter) SendVoteRequest(server *Server, peer *Peer, req *Reque
defer httpResp.Body.Close()
resp := &RequestVoteResponse{}
if _, err = resp.decode(httpResp.Body); err != nil && err != io.EOF {
if _, err = resp.Decode(httpResp.Body); err != nil && err != io.EOF {
traceln("transporter.rv.decoding.error:", err)
return nil
}
@ -162,13 +162,13 @@ func (t *HTTPTransporter) appendEntriesHandler(server *Server) http.HandlerFunc
traceln(server.Name(), "RECV /appendEntries")
req := &AppendEntriesRequest{}
if _, err := req.decode(r.Body); err != nil {
if _, err := req.Decode(r.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
return
}
resp := server.AppendEntries(req)
if _, err := resp.encode(w); err != nil {
if _, err := resp.Encode(w); err != nil {
http.Error(w, "", http.StatusInternalServerError)
return
}
@ -181,13 +181,13 @@ func (t *HTTPTransporter) requestVoteHandler(server *Server) http.HandlerFunc {
traceln(server.Name(), "RECV /requestVote")
req := &RequestVoteRequest{}
if _, err := req.decode(r.Body); err != nil {
if _, err := req.Decode(r.Body); err != nil {
http.Error(w, "", http.StatusBadRequest)
return
}
resp := server.RequestVote(req)
if _, err := resp.encode(w); err != nil {
if _, err := resp.Encode(w); err != nil {
http.Error(w, "", http.StatusInternalServerError)
return
}

30
third_party/github.com/coreos/go-raft/log.go vendored
View File

@ -180,26 +180,23 @@ func (l *Log) open(path string) error {
}
break
}
// Append entry.
l.entries = append(l.entries, entry)
if entry.Index <= l.commitIndex {
command, err := newCommand(entry.CommandName, entry.Command)
if err != nil {
continue
if entry.Index > l.startIndex {
// Append entry.
l.entries = append(l.entries, entry)
if entry.Index <= l.commitIndex {
command, err := newCommand(entry.CommandName, entry.Command)
if err != nil {
continue
}
l.ApplyFunc(command)
}
l.ApplyFunc(command)
debugln("open.log.append log index ", entry.Index)
}
debugln("open.log.append log index ", entry.Index)
readBytes += int64(n)
}
l.results = make([]*logResult, len(l.entries))
l.compact(l.startIndex, l.startTerm)
debugln("open.log.recovery number of log ", len(l.entries))
return nil
}
@ -273,6 +270,8 @@ func (l *Log) getEntriesAfter(index uint64, maxLogEntriesPerRequest uint64) ([]*
entries := l.entries[index-l.startIndex:]
length := len(entries)
traceln("log.entriesAfter: startIndex:", l.startIndex, " lenght", len(l.entries))
if uint64(length) < maxLogEntriesPerRequest {
// Determine the term at the given entry and return a subslice.
return entries, l.entries[index-1-l.startIndex].Term
@ -353,7 +352,10 @@ func (l *Log) lastInfo() (index uint64, term uint64) {
func (l *Log) updateCommitIndex(index uint64) {
l.mutex.Lock()
defer l.mutex.Unlock()
l.commitIndex = index
if index > l.commitIndex {
l.commitIndex = index
}
debugln("update.commit.index ", index)
}
// Updates the commit index and writes entries after that index to the stable storage.

6
third_party/github.com/coreos/go-raft/peer.go vendored
View File

@ -255,6 +255,12 @@ func (p *Peer) sendSnapshotRecoveryRequest() {
req := newSnapshotRecoveryRequest(p.server.name, p.server.lastSnapshot)
debugln("peer.snap.recovery.send: ", p.Name)
resp := p.server.Transporter().SendSnapshotRecoveryRequest(p.server, p, req)
if resp == nil {
debugln("peer.snap.recovery.timeout: ", p.Name)
return
}
if resp.Success {
p.prevLogIndex = req.LastIndex
} else {

4
third_party/github.com/coreos/go-raft/request_vote_request.go vendored
View File

@ -28,7 +28,7 @@ func newRequestVoteRequest(term uint64, candidateName string, lastLogIndex uint6
// Encodes the RequestVoteRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *RequestVoteRequest) encode(w io.Writer) (int, error) {
func (req *RequestVoteRequest) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoRequestVoteRequest{
Term: proto.Uint64(req.Term),
LastLogIndex: proto.Uint64(req.LastLogIndex),
@ -45,7 +45,7 @@ func (req *RequestVoteRequest) encode(w io.Writer) (int, error) {
// Decodes the RequestVoteRequest from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *RequestVoteRequest) decode(r io.Reader) (int, error) {
func (req *RequestVoteRequest) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

4
third_party/github.com/coreos/go-raft/request_vote_response.go vendored
View File

@ -24,7 +24,7 @@ func newRequestVoteResponse(term uint64, voteGranted bool) *RequestVoteResponse
// Encodes the RequestVoteResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (resp *RequestVoteResponse) encode(w io.Writer) (int, error) {
func (resp *RequestVoteResponse) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoRequestVoteResponse{
Term: proto.Uint64(resp.Term),
VoteGranted: proto.Bool(resp.VoteGranted),
@ -40,7 +40,7 @@ func (resp *RequestVoteResponse) encode(w io.Writer) (int, error) {
// Decodes the RequestVoteResponse from a buffer. Returns the number of bytes read and
// any error that occurs.
func (resp *RequestVoteResponse) decode(r io.Reader) (int, error) {
func (resp *RequestVoteResponse) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

18
third_party/github.com/coreos/go-raft/server.go vendored
View File

@ -80,6 +80,8 @@ type Server struct {
lastSnapshot *Snapshot
stateMachine StateMachine
maxLogEntriesPerRequest uint64
connectionString string
}
// An event to be processed by the server's event loop.
@ -96,7 +98,7 @@ type event struct {
//------------------------------------------------------------------------------
// Creates a new server with a log at the given path.
func NewServer(name string, path string, transporter Transporter, stateMachine StateMachine, context interface{}) (*Server, error) {
func NewServer(name string, path string, transporter Transporter, stateMachine StateMachine, context interface{}, connectionString string) (*Server, error) {
if name == "" {
return nil, errors.New("raft.Server: Name cannot be blank")
}
@ -117,6 +119,7 @@ func NewServer(name string, path string, transporter Transporter, stateMachine S
electionTimeout: DefaultElectionTimeout,
heartbeatTimeout: DefaultHeartbeatTimeout,
maxLogEntriesPerRequest: MaxLogEntriesPerRequest,
connectionString: connectionString,
}
// Setup apply function.
@ -1009,10 +1012,17 @@ func (s *Server) TakeSnapshot() error {
state = []byte{0}
}
var peers []*Peer
peers := make([]*Peer, len(s.peers)+1)
i := 0
for _, peer := range s.peers {
peers = append(peers, peer.clone())
peers[i] = peer.clone()
i++
}
peers[i] = &Peer{
Name: s.Name(),
ConnectionString: s.connectionString,
}
s.currentSnapshot = &Snapshot{lastIndex, lastTerm, peers, state, path}
@ -1253,7 +1263,7 @@ func (s *Server) readConf() error {
return err
}
s.log.commitIndex = conf.CommitIndex
s.log.updateCommitIndex(conf.CommitIndex)
return nil
}

2
third_party/github.com/coreos/go-raft/server_test.go vendored
View File

@ -428,7 +428,7 @@ func TestServerRecoverFromPreviousLogAndConf(t *testing.T) {
for _, name := range names {
server := servers[name]
if server.CommitIndex() != 17 {
t.Fatalf("%s commitIndex is invalid [%d/%d]", name, server.CommitIndex(), 16)
t.Fatalf("%s commitIndex is invalid [%d/%d]", name, server.CommitIndex(), 17)
}
server.Stop()
}

4
third_party/github.com/coreos/go-raft/snapshot_recovery_request.go vendored
View File

@ -35,7 +35,7 @@ func newSnapshotRecoveryRequest(leaderName string, snapshot *Snapshot) *Snapshot
// Encodes the SnapshotRecoveryRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRecoveryRequest) encode(w io.Writer) (int, error) {
func (req *SnapshotRecoveryRequest) Encode(w io.Writer) (int, error) {
protoPeers := make([]*protobuf.ProtoSnapshotRecoveryRequest_ProtoPeer, len(req.Peers))
@ -63,7 +63,7 @@ func (req *SnapshotRecoveryRequest) encode(w io.Writer) (int, error) {
// Decodes the SnapshotRecoveryRequest from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *SnapshotRecoveryRequest) decode(r io.Reader) (int, error) {
func (req *SnapshotRecoveryRequest) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

4
third_party/github.com/coreos/go-raft/snapshot_recovery_response.go vendored
View File

@ -31,7 +31,7 @@ func newSnapshotRecoveryResponse(term uint64, success bool, commitIndex uint64)
// Encodes the SnapshotRecoveryResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRecoveryResponse) encode(w io.Writer) (int, error) {
func (req *SnapshotRecoveryResponse) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoSnapshotRecoveryResponse{
Term: proto.Uint64(req.Term),
Success: proto.Bool(req.Success),
@ -47,7 +47,7 @@ func (req *SnapshotRecoveryResponse) encode(w io.Writer) (int, error) {
// Decodes the SnapshotRecoveryResponse from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *SnapshotRecoveryResponse) decode(r io.Reader) (int, error) {
func (req *SnapshotRecoveryResponse) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

4
third_party/github.com/coreos/go-raft/snapshot_request.go vendored
View File

@ -31,7 +31,7 @@ func newSnapshotRequest(leaderName string, snapshot *Snapshot) *SnapshotRequest
// Encodes the SnapshotRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRequest) encode(w io.Writer) (int, error) {
func (req *SnapshotRequest) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoSnapshotRequest{
LeaderName: proto.String(req.LeaderName),
LastIndex: proto.Uint64(req.LastIndex),
@ -47,7 +47,7 @@ func (req *SnapshotRequest) encode(w io.Writer) (int, error) {
// Decodes the SnapshotRequest from a buffer. Returns the number of bytes read and
// any error that occurs.
func (req *SnapshotRequest) decode(r io.Reader) (int, error) {
func (req *SnapshotRequest) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

4
third_party/github.com/coreos/go-raft/snapshot_response.go vendored
View File

@ -27,7 +27,7 @@ func newSnapshotResponse(success bool) *SnapshotResponse {
// Encodes the SnapshotResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (resp *SnapshotResponse) encode(w io.Writer) (int, error) {
func (resp *SnapshotResponse) Encode(w io.Writer) (int, error) {
pb := &protobuf.ProtoSnapshotResponse{
Success: proto.Bool(resp.Success),
}
@ -41,7 +41,7 @@ func (resp *SnapshotResponse) encode(w io.Writer) (int, error) {
// Decodes the SnapshotResponse from a buffer. Returns the number of bytes read and
// any error that occurs.
func (resp *SnapshotResponse) decode(r io.Reader) (int, error) {
func (resp *SnapshotResponse) Decode(r io.Reader) (int, error) {
data, err := ioutil.ReadAll(r)
if err != nil {

4
third_party/github.com/coreos/go-raft/test.go vendored
View File

@ -65,12 +65,12 @@ func newTestServer(name string, transporter Transporter) *Server {
if err := os.MkdirAll(p, 0644); err != nil {
panic(err.Error())
}
server, _ := NewServer(name, p, transporter, nil, nil)
server, _ := NewServer(name, p, transporter, nil, nil, "")
return server
}
func newTestServerWithPath(name string, transporter Transporter, p string) *Server {
server, _ := NewServer(name, p, transporter, nil, nil)
server, _ := NewServer(name, p, transporter, nil, nil, "")
return server
}

4
transporter.go
View File

@ -66,7 +66,7 @@ func (t *transporter) SendAppendEntriesRequest(server *raft.Server, peer *raft.P
debugf("Send LogEntries to %s ", u)
thisPeerStats, ok := r.peersStats[peer.Name]
thisPeerStats, ok := r.peersStats.Peers[peer.Name]
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 {
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)
}
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) {
if err != nil {
fatal(err)