package serf
|
|
import (
|
"bytes"
|
"time"
|
|
"github.com/armon/go-metrics"
|
"github.com/hashicorp/go-msgpack/codec"
|
"github.com/hashicorp/memberlist"
|
"basic.com/valib/serf.git/coordinate"
|
)
|
|
// pingDelegate is notified when memberlist successfully completes a direct ping
|
// of a peer node. We use this to update our estimated network coordinate, as
|
// well as cache the coordinate of the peer.
|
type pingDelegate struct {
|
serf *Serf
|
}
|
|
const (
|
// PingVersion is an internal version for the ping message, above the normal
|
// versioning we get from the protocol version. This enables small updates
|
// to the ping message without a full protocol bump.
|
PingVersion = 1
|
)
|
|
// AckPayload is called to produce a payload to send back in response to a ping
|
// request.
|
func (p *pingDelegate) AckPayload() []byte {
|
var buf bytes.Buffer
|
|
// The first byte is the version number, forming a simple header.
|
version := []byte{PingVersion}
|
buf.Write(version)
|
|
// The rest of the message is the serialized coordinate.
|
enc := codec.NewEncoder(&buf, &codec.MsgpackHandle{})
|
if err := enc.Encode(p.serf.coordClient.GetCoordinate()); err != nil {
|
p.serf.logger.Printf("[ERR] serf: Failed to encode coordinate: %v\n", err)
|
}
|
return buf.Bytes()
|
}
|
|
// NotifyPingComplete is called when this node successfully completes a direct ping
|
// of a peer node.
|
func (p *pingDelegate) NotifyPingComplete(other *memberlist.Node, rtt time.Duration, payload []byte) {
|
if payload == nil || len(payload) == 0 {
|
return
|
}
|
|
// Verify ping version in the header.
|
version := payload[0]
|
if version != PingVersion {
|
p.serf.logger.Printf("[ERR] serf: Unsupported ping version: %v", version)
|
return
|
}
|
|
// Process the remainder of the message as a coordinate.
|
r := bytes.NewReader(payload[1:])
|
dec := codec.NewDecoder(r, &codec.MsgpackHandle{})
|
var coord coordinate.Coordinate
|
if err := dec.Decode(&coord); err != nil {
|
p.serf.logger.Printf("[ERR] serf: Failed to decode coordinate from ping: %v", err)
|
return
|
}
|
|
// Apply the update.
|
before := p.serf.coordClient.GetCoordinate()
|
after, err := p.serf.coordClient.Update(other.Name, &coord, rtt)
|
if err != nil {
|
metrics.IncrCounter([]string{"serf", "coordinate", "rejected"}, 1)
|
p.serf.logger.Printf("[TRACE] serf: Rejected coordinate from %s: %v\n",
|
other.Name, err)
|
return
|
}
|
|
// Publish some metrics to give us an idea of how much we are
|
// adjusting each time we update.
|
d := float32(before.DistanceTo(after).Seconds() * 1.0e3)
|
metrics.AddSample([]string{"serf", "coordinate", "adjustment-ms"}, d)
|
|
// Cache the coordinate for the other node, and add our own
|
// to the cache as well since it just got updated. This lets
|
// users call GetCachedCoordinate with our node name, which is
|
// more friendly.
|
p.serf.coordCacheLock.Lock()
|
p.serf.coordCache[other.Name] = &coord
|
p.serf.coordCache[p.serf.config.NodeName] = p.serf.coordClient.GetCoordinate()
|
p.serf.coordCacheLock.Unlock()
|
}
|
