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 3ea1c7d
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,158 @@ | ||
| package rapidpro | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "maps" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types" | ||
| "github.com/gomodule/redigo/redis" | ||
| "github.com/nyaruka/courier" | ||
| "github.com/nyaruka/gocommon/aws/cwatch" | ||
| ) | ||
|
|
||
| type CountByType map[courier.ChannelType]int | ||
|
|
||
| func (c CountByType) Metrics(name string) []types.MetricDatum { | ||
| m := make([]types.MetricDatum, 0, len(c)) | ||
| for typ, count := range c { | ||
| m = append(m, cwatch.Datum(name, float64(count), types.StandardUnitCount, cwatch.Dimension("ChannelType", string(typ)))) | ||
| } | ||
| return m | ||
| } | ||
|
|
||
| type DurationByType map[courier.ChannelType]time.Duration | ||
|
|
||
| type Stats struct { | ||
| IncomingRequests CountByType // number of handler requests | ||
| IncomingMessages CountByType // number of messages received | ||
| IncomingStatuses CountByType // number of status updates received | ||
| IncomingEvents CountByType // number of other events received | ||
| IncomingIgnored CountByType // number of requests ignored | ||
| IncomingDuration DurationByType // total time spent handling requests | ||
|
|
||
| OutgoingSends CountByType // number of sends that succeeded | ||
| OutgoingErrors CountByType // number of sends that errored | ||
| OutgoingDuration DurationByType // total time spent sending messages | ||
|
|
||
| ContactsCreated int | ||
|
|
||
| DBWaitDuration time.Duration | ||
| RedisWaitDuration time.Duration | ||
| } | ||
|
|
||
| func newStats() *Stats { | ||
| return &Stats{ | ||
| IncomingRequests: make(CountByType), | ||
| IncomingMessages: make(CountByType), | ||
| IncomingStatuses: make(CountByType), | ||
| IncomingEvents: make(CountByType), | ||
| IncomingIgnored: make(CountByType), | ||
| IncomingDuration: make(DurationByType), | ||
|
|
||
| OutgoingSends: make(CountByType), | ||
| OutgoingErrors: make(CountByType), | ||
| OutgoingDuration: make(DurationByType), | ||
|
|
||
| ContactsCreated: 0, | ||
| } | ||
| } | ||
|
|
||
| func (s *Stats) reset(db sql.DBStats, rp redis.PoolStats) { | ||
| clear(s.IncomingRequests) | ||
| clear(s.IncomingMessages) | ||
| clear(s.IncomingStatuses) | ||
| clear(s.IncomingEvents) | ||
| clear(s.IncomingIgnored) | ||
| clear(s.IncomingDuration) | ||
|
|
||
| clear(s.OutgoingSends) | ||
| clear(s.OutgoingErrors) | ||
| clear(s.OutgoingDuration) | ||
|
|
||
| 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) RecordIncoming(typ courier.ChannelType, evts []courier.Event, d time.Duration) { | ||
| c.mutex.Lock() | ||
| c.stats.IncomingRequests[typ]++ | ||
|
|
||
| for _, e := range evts { | ||
| switch e.(type) { | ||
| case courier.MsgIn: | ||
| c.stats.IncomingMessages[typ]++ | ||
| case courier.StatusUpdate: | ||
| c.stats.IncomingStatuses[typ]++ | ||
| case courier.ChannelEvent: | ||
| c.stats.IncomingEvents[typ]++ | ||
| } | ||
| } | ||
| if len(evts) == 0 { | ||
| c.stats.IncomingIgnored[typ]++ | ||
| } | ||
|
|
||
| c.stats.IncomingDuration[typ] += d | ||
| c.mutex.Unlock() | ||
| } | ||
|
|
||
| func (c *StatsCollector) RecordOutgoing(typ courier.ChannelType, success bool, d time.Duration) { | ||
| c.mutex.Lock() | ||
| if success { | ||
| c.stats.OutgoingSends[typ]++ | ||
| } else { | ||
| c.stats.OutgoingErrors[typ]++ | ||
| } | ||
| c.stats.OutgoingDuration[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, | ||
|
|
||
| IncomingRequests: maps.Clone(c.stats.IncomingRequests), | ||
| IncomingMessages: maps.Clone(c.stats.IncomingMessages), | ||
| IncomingStatuses: maps.Clone(c.stats.IncomingStatuses), | ||
| IncomingEvents: maps.Clone(c.stats.IncomingEvents), | ||
| IncomingIgnored: maps.Clone(c.stats.IncomingIgnored), | ||
| IncomingDuration: maps.Clone(c.stats.IncomingDuration), | ||
|
|
||
| OutgoingSends: maps.Clone(c.stats.OutgoingSends), | ||
| OutgoingErrors: maps.Clone(c.stats.OutgoingErrors), | ||
| OutgoingDuration: maps.Clone(c.stats.OutgoingDuration), | ||
|
|
||
| // 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.