Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor metrics so that everything is sent from Heartbeat in the bac…
…kend
- Loading branch information
1 parent
68d55f9
commit f191230
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| package rapidpro | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "maps" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/gomodule/redigo/redis" | ||
| "github.com/nyaruka/courier" | ||
| ) | ||
|
|
||
| type Stats struct { | ||
| ReceiveRequests map[courier.ChannelType]int // number of handler requests | ||
| ReceiveMessages map[courier.ChannelType]int // number of messages received | ||
| ReceiveStatuses map[courier.ChannelType]int // number of status updates received | ||
| ReceiveEvents map[courier.ChannelType]int // number of other events received | ||
| ReceiveIgnored map[courier.ChannelType]int // number of requests ignored | ||
| ReceiveDuration map[courier.ChannelType]time.Duration | ||
|
|
||
| SendSuccesses map[courier.ChannelType]int // number of sends that succeeded | ||
| SendErrors map[courier.ChannelType]int // number of sends that errored | ||
| SendDuration map[courier.ChannelType]time.Duration | ||
|
|
||
| ContactsCreated int | ||
|
|
||
| DBWaitDuration time.Duration | ||
| RedisWaitDuration time.Duration | ||
| } | ||
|
|
||
| func newStats() *Stats { | ||
| return &Stats{ | ||
| ReceiveRequests: make(map[courier.ChannelType]int), | ||
| ReceiveMessages: make(map[courier.ChannelType]int), | ||
| ReceiveStatuses: make(map[courier.ChannelType]int), | ||
| ReceiveEvents: make(map[courier.ChannelType]int), | ||
| ReceiveIgnored: make(map[courier.ChannelType]int), | ||
| ReceiveDuration: make(map[courier.ChannelType]time.Duration), | ||
|
|
||
| SendSuccesses: make(map[courier.ChannelType]int), | ||
| SendErrors: make(map[courier.ChannelType]int), | ||
| SendDuration: make(map[courier.ChannelType]time.Duration), | ||
|
|
||
| ContactsCreated: 0, | ||
| } | ||
| } | ||
|
|
||
| func (s *Stats) reset(db sql.DBStats, rp redis.PoolStats) { | ||
| clear(s.ReceiveRequests) | ||
| clear(s.ReceiveMessages) | ||
| clear(s.ReceiveStatuses) | ||
| clear(s.ReceiveEvents) | ||
| clear(s.ReceiveIgnored) | ||
| clear(s.ReceiveDuration) | ||
|
|
||
| clear(s.SendSuccesses) | ||
| clear(s.SendErrors) | ||
| clear(s.SendDuration) | ||
|
|
||
| s.ContactsCreated = 0 | ||
|
|
||
| s.DBWaitDuration = db.WaitDuration | ||
| s.RedisWaitDuration = rp.WaitDuration | ||
| } | ||
|
|
||
| // StatsCollector provides threadsafe stats collection | ||
| type StatsCollector struct { | ||
| mutex sync.Mutex | ||
| stats *Stats | ||
| } | ||
|
|
||
| // NewStatsCollector creates a new stats collector | ||
| func NewStatsCollector() *StatsCollector { | ||
| return &StatsCollector{stats: newStats()} | ||
| } | ||
|
|
||
| func (c *StatsCollector) RecordReceive(typ courier.ChannelType, evts []courier.Event, d time.Duration) { | ||
| c.mutex.Lock() | ||
| c.stats.ReceiveRequests[typ]++ | ||
|
|
||
| for _, e := range evts { | ||
| switch e.(type) { | ||
| case courier.MsgIn: | ||
| c.stats.ReceiveMessages[typ]++ | ||
| case courier.StatusUpdate: | ||
| c.stats.ReceiveStatuses[typ]++ | ||
| case courier.ChannelEvent: | ||
| c.stats.ReceiveEvents[typ]++ | ||
| } | ||
| } | ||
| if len(evts) == 0 { | ||
| c.stats.ReceiveIgnored[typ]++ | ||
| } | ||
|
|
||
| c.stats.ReceiveDuration[typ] += d | ||
| c.mutex.Unlock() | ||
| } | ||
|
|
||
| func (c *StatsCollector) RecordSend(typ courier.ChannelType, success bool, d time.Duration) { | ||
| c.mutex.Lock() | ||
| if success { | ||
| c.stats.SendSuccesses[typ]++ | ||
| } else { | ||
| c.stats.SendErrors[typ]++ | ||
| } | ||
| c.stats.SendDuration[typ] += d | ||
| c.mutex.Unlock() | ||
| } | ||
|
|
||
| func (c *StatsCollector) RecordContactCreated() { | ||
| c.mutex.Lock() | ||
| c.stats.ContactsCreated++ | ||
| c.mutex.Unlock() | ||
| } | ||
|
|
||
| // Stats returns the stats for the period since the last call | ||
| func (c *StatsCollector) Stats(db sql.DBStats, rp redis.PoolStats) *Stats { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
|
|
||
| stats := &Stats{ | ||
| ContactsCreated: c.stats.ContactsCreated, | ||
|
|
||
| ReceiveRequests: maps.Clone(c.stats.ReceiveRequests), | ||
| ReceiveMessages: maps.Clone(c.stats.ReceiveMessages), | ||
| ReceiveStatuses: maps.Clone(c.stats.ReceiveStatuses), | ||
| ReceiveEvents: maps.Clone(c.stats.ReceiveEvents), | ||
| ReceiveIgnored: maps.Clone(c.stats.ReceiveIgnored), | ||
| ReceiveDuration: maps.Clone(c.stats.ReceiveDuration), | ||
|
|
||
| SendSuccesses: maps.Clone(c.stats.SendSuccesses), | ||
| SendErrors: maps.Clone(c.stats.SendErrors), | ||
| SendDuration: maps.Clone(c.stats.SendDuration), | ||
|
|
||
| // both sqlx and redis provide wait stats which are cummulative that we need to convert into increments by | ||
| // tracking their previous values | ||
| DBWaitDuration: db.WaitDuration - c.stats.DBWaitDuration, | ||
| RedisWaitDuration: rp.WaitDuration - c.stats.RedisWaitDuration, | ||
| } | ||
|
|
||
| c.stats.reset(db, rp) | ||
|
|
||
| return stats | ||
| } | ||
Oops, something went wrong.