You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

300 lines
7.3 KiB

package eventmeter
import (
"encoding/json"
"fmt"
"sync"
"time"
"github.com/gorilla/websocket"
metrics "github.com/rcrowley/go-metrics"
events "github.com/tendermint/go-events"
client "github.com/tendermint/go-rpc/client"
log15 "github.com/tendermint/log15"
)
// Log allows you to set your own logger.
var Log log15.Logger
//------------------------------------------------------
// Generic system to subscribe to events and record their frequency
//------------------------------------------------------
//------------------------------------------------------
// Meter for a particular event
// Closure to enable side effects from receiving an event
type EventCallbackFunc func(em *EventMetric, data events.EventData)
// Metrics for a given event
type EventMetric struct {
ID string `json:"id"`
Started time.Time `json:"start_time"`
LastHeard time.Time `json:"last_heard"`
MinDuration int64 `json:"min_duration"`
MaxDuration int64 `json:"max_duration"`
// tracks event count and rate
meter metrics.Meter
// filled in from the Meter
Count int64 `json:"count"`
Rate1 float64 `json:"rate_1" wire:"unsafe"`
Rate5 float64 `json:"rate_5" wire:"unsafe"`
Rate15 float64 `json:"rate_15" wire:"unsafe"`
RateMean float64 `json:"rate_mean" wire:"unsafe"`
// so the event can have effects in the event-meter's consumer.
// runs in a go routine
callback EventCallbackFunc
}
func (metric *EventMetric) Copy() *EventMetric {
metric2 := *metric
metric2.meter = metric.meter.Snapshot()
return &metric2
}
// called on GetMetric
func (metric *EventMetric) fillMetric() *EventMetric {
metric.Count = metric.meter.Count()
metric.Rate1 = metric.meter.Rate1()
metric.Rate5 = metric.meter.Rate5()
metric.Rate15 = metric.meter.Rate15()
metric.RateMean = metric.meter.RateMean()
return metric
}
//------------------------------------------------------
// Websocket client and event meter for many events
const maxPingsPerPong = 30 // if we haven't received a pong in this many attempted pings we kill the conn
// Get the eventID and data out of the raw json received over the go-rpc websocket
type EventUnmarshalFunc func(b json.RawMessage) (string, events.EventData, error)
// Closure to enable side effects from receiving a pong
type LatencyCallbackFunc func(meanLatencyNanoSeconds float64)
// Closure to notify consumer that the connection died
type DisconnectCallbackFunc func()
// Each node gets an event meter to track events for that node
type EventMeter struct {
wsc *client.WSClient
mtx sync.Mutex
events map[string]*EventMetric
// to record ws latency
timer metrics.Timer
lastPing time.Time
receivedPong bool
latencyCallback LatencyCallbackFunc
disconnectCallback DisconnectCallbackFunc
unmarshalEvent EventUnmarshalFunc
quit chan struct{}
}
func NewEventMeter(addr string, unmarshalEvent EventUnmarshalFunc) *EventMeter {
em := &EventMeter{
wsc: client.NewWSClient(addr, "/websocket"),
events: make(map[string]*EventMetric),
timer: metrics.NewTimer(),
receivedPong: true,
unmarshalEvent: unmarshalEvent,
quit: make(chan struct{}),
}
return em
}
func (em *EventMeter) String() string {
return em.wsc.Address
}
func (em *EventMeter) Start() error {
if _, err := em.wsc.Start(); err != nil {
return err
}
em.wsc.Conn.SetPongHandler(func(m string) error {
// NOTE: https://github.com/gorilla/websocket/issues/97
em.mtx.Lock()
defer em.mtx.Unlock()
em.receivedPong = true
em.timer.UpdateSince(em.lastPing)
if em.latencyCallback != nil {
go em.latencyCallback(em.timer.Mean())
}
return nil
})
go em.receiveRoutine()
return nil
}
func (em *EventMeter) Stop() {
<-em.quit
em.RegisterDisconnectCallback(nil) // so we don't try and reconnect
em.wsc.Stop() // close(wsc.Quit)
}
func (em *EventMeter) StopAndReconnect() {
em.wsc.Stop()
em.mtx.Lock()
defer em.mtx.Unlock()
if em.disconnectCallback != nil {
go em.disconnectCallback()
}
}
func (em *EventMeter) Subscribe(eventID string, cb EventCallbackFunc) error {
em.mtx.Lock()
defer em.mtx.Unlock()
if _, ok := em.events[eventID]; ok {
return fmt.Errorf("subscribtion already exists")
}
if err := em.wsc.Subscribe(eventID); err != nil {
return err
}
metric := &EventMetric{
ID: eventID,
Started: time.Now(),
MinDuration: 1 << 62,
meter: metrics.NewMeter(),
callback: cb,
}
em.events[eventID] = metric
return nil
}
func (em *EventMeter) Unsubscribe(eventID string) error {
em.mtx.Lock()
defer em.mtx.Unlock()
if err := em.wsc.Unsubscribe(eventID); err != nil {
return err
}
// XXX: should we persist or save this info first?
delete(em.events, eventID)
return nil
}
// Fill in the latest data for an event and return a copy
func (em *EventMeter) GetMetric(eventID string) (*EventMetric, error) {
em.mtx.Lock()
defer em.mtx.Unlock()
metric, ok := em.events[eventID]
if !ok {
return nil, fmt.Errorf("Unknown event %s", eventID)
}
return metric.fillMetric().Copy(), nil
}
// Return the average latency over the websocket
func (em *EventMeter) Latency() float64 {
em.mtx.Lock()
defer em.mtx.Unlock()
return em.timer.Mean()
}
func (em *EventMeter) RegisterLatencyCallback(f LatencyCallbackFunc) {
em.mtx.Lock()
defer em.mtx.Unlock()
em.latencyCallback = f
}
func (em *EventMeter) RegisterDisconnectCallback(f DisconnectCallbackFunc) {
em.mtx.Lock()
defer em.mtx.Unlock()
em.disconnectCallback = f
}
//------------------------------------------------------
func (em *EventMeter) receiveRoutine() {
pingTime := time.Second * 1
pingTicker := time.NewTicker(pingTime)
pingAttempts := 0 // if this hits maxPingsPerPong we kill the conn
var err error
for {
select {
case <-pingTicker.C:
if pingAttempts, err = em.pingForLatency(pingAttempts); err != nil {
Log.Error("Failed to write ping message on websocket", err)
em.StopAndReconnect()
return
} else if pingAttempts >= maxPingsPerPong {
Log.Error(fmt.Sprintf("Have not received a pong in %v", time.Duration(pingAttempts)*pingTime))
em.StopAndReconnect()
return
}
case r := <-em.wsc.ResultsCh:
if r == nil {
em.StopAndReconnect()
return
}
eventID, data, err := em.unmarshalEvent(r)
if err != nil {
Log.Error(err.Error())
continue
}
if eventID != "" {
em.updateMetric(eventID, data)
}
case <-em.wsc.Quit:
em.StopAndReconnect()
return
case <-em.quit:
return
}
}
}
func (em *EventMeter) pingForLatency(pingAttempts int) (int, error) {
em.mtx.Lock()
defer em.mtx.Unlock()
// ping to record latency
if !em.receivedPong {
return pingAttempts + 1, nil
}
em.lastPing = time.Now()
em.receivedPong = false
err := em.wsc.Conn.WriteMessage(websocket.PingMessage, []byte{})
if err != nil {
return pingAttempts, err
}
return 0, nil
}
func (em *EventMeter) updateMetric(eventID string, data events.EventData) {
em.mtx.Lock()
defer em.mtx.Unlock()
metric, ok := em.events[eventID]
if !ok {
// we already unsubscribed, or got an unexpected event
return
}
last := metric.LastHeard
metric.LastHeard = time.Now()
metric.meter.Mark(1)
dur := int64(metric.LastHeard.Sub(last))
if dur < metric.MinDuration {
metric.MinDuration = dur
}
if !last.IsZero() && dur > metric.MaxDuration {
metric.MaxDuration = dur
}
if metric.callback != nil {
go metric.callback(metric.Copy(), data)
}
}