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.
330 lines
8.0 KiB
330 lines
8.0 KiB
package eventmeter
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/go-kit/kit/log"
|
|
"github.com/gorilla/websocket"
|
|
"github.com/pkg/errors"
|
|
metrics "github.com/rcrowley/go-metrics"
|
|
events "github.com/tendermint/go-events"
|
|
client "github.com/tendermint/go-rpc/client"
|
|
)
|
|
|
|
//------------------------------------------------------
|
|
// 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{}
|
|
|
|
logger log.Logger
|
|
}
|
|
|
|
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,
|
|
logger: log.NewNopLogger(),
|
|
}
|
|
return em
|
|
}
|
|
|
|
// SetLogger lets you set your own logger
|
|
func (em *EventMeter) SetLogger(l log.Logger) {
|
|
em.logger = l
|
|
}
|
|
|
|
func (em *EventMeter) String() string {
|
|
return em.wsc.Address
|
|
}
|
|
|
|
func (em *EventMeter) Start() error {
|
|
if _, err := em.wsc.Reset(); err != nil {
|
|
return err
|
|
}
|
|
|
|
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
|
|
})
|
|
|
|
em.quit = make(chan struct{})
|
|
go em.receiveRoutine()
|
|
|
|
return em.resubscribe()
|
|
}
|
|
|
|
// Stop stops the EventMeter.
|
|
func (em *EventMeter) Stop() {
|
|
close(em.quit)
|
|
|
|
if em.wsc.IsRunning() {
|
|
em.wsc.Stop()
|
|
}
|
|
}
|
|
|
|
// StopAndCallDisconnectCallback stops the EventMeter and calls
|
|
// disconnectCallback if present.
|
|
func (em *EventMeter) StopAndCallDisconnectCallback() {
|
|
if em.wsc.IsRunning() {
|
|
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) resubscribe() error {
|
|
for eventID, _ := range em.events {
|
|
if err := em.wsc.Subscribe(eventID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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 {
|
|
em.logger.Log("err", errors.Wrap(err, "failed to write ping message on websocket"))
|
|
em.StopAndCallDisconnectCallback()
|
|
return
|
|
} else if pingAttempts >= maxPingsPerPong {
|
|
em.logger.Log("err", errors.Errorf("Have not received a pong in %v", time.Duration(pingAttempts)*pingTime))
|
|
em.StopAndCallDisconnectCallback()
|
|
return
|
|
}
|
|
case r := <-em.wsc.ResultsCh:
|
|
if r == nil {
|
|
em.logger.Log("err", errors.New("Expected some event, received nil"))
|
|
em.StopAndCallDisconnectCallback()
|
|
return
|
|
}
|
|
eventID, data, err := em.unmarshalEvent(r)
|
|
if err != nil {
|
|
em.logger.Log("err", errors.Wrap(err, "failed to unmarshal event"))
|
|
continue
|
|
}
|
|
if eventID != "" {
|
|
em.updateMetric(eventID, data)
|
|
}
|
|
case <-em.wsc.Quit:
|
|
em.logger.Log("err", errors.New("WSClient closed unexpectedly"))
|
|
em.StopAndCallDisconnectCallback()
|
|
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)
|
|
}
|
|
}
|