From 6116ccc4aa54937690887077057c0bfb939dd296 Mon Sep 17 00:00:00 2001 From: Daniel Ponte Date: Wed, 23 Oct 2024 08:55:19 -0400 Subject: [PATCH] WIP --- internal/trending/item.go | 24 ++--- internal/trending/score.go | 16 ++-- internal/trending/trending.go | 68 +++++++------- pkg/gordio/server/server.go | 4 + pkg/gordio/sinks/alerting/alerting.go | 122 ++++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 54 deletions(-) create mode 100644 pkg/gordio/sinks/alerting/alerting.go diff --git a/internal/trending/item.go b/internal/trending/item.go index 534ef41..02517e5 100644 --- a/internal/trending/item.go +++ b/internal/trending/item.go @@ -5,23 +5,23 @@ import ( "time" ) -type item struct { +type item[K comparable] struct { eventSeries TimeSeries maxSeries SlidingWindow max float64 maxTime time.Time - options *options + options *options[K] // TODO: move outside of item because it's the same for all items defaultExpectation float64 defaultHourlyCount float64 } -func newItem(id string, options *options) *item { +func newItem[K comparable](id K, options *options[K]) *item[K] { defaultHourlyCount := float64(options.baseCount) * float64(options.storageDuration/time.Hour) defaultExpectation := float64(options.baseCount) / float64(time.Hour/options.recentDuration) - return &item{ + return &item[K]{ eventSeries: options.creator(id), maxSeries: options.slidingWindowCreator(id), options: options, @@ -31,10 +31,10 @@ func newItem(id string, options *options) *item { } } -func (i *item) score() score { +func (i *item[K]) score() score[K] { recentCount, count := i.computeCounts() if recentCount < i.options.countThreshold { - return score{} + return score[K]{} } if recentCount == count { // we see this for the first time so there is no historical data @@ -60,7 +60,7 @@ func (i *item) score() score { mixedScore := 0.5 * (klScore + i.max) - return score{ + return score[K]{ Score: mixedScore, Probability: probability, Expectation: expectation, @@ -69,27 +69,27 @@ func (i *item) score() score { } } -func (i *item) computeCounts() (float64, float64) { +func (i *item[K]) computeCounts() (float64, float64) { now := time.Now() totalCount, _ := i.eventSeries.Range(now.Add(-i.options.storageDuration), now) count, _ := i.eventSeries.Range(now.Add(-i.options.recentDuration), now) return count, totalCount } -func (i *item) computeRecentMax() float64 { +func (i *item[K]) computeRecentMax() float64 { return i.maxSeries.Max() } -func (i *item) decayMax() { +func (i *item[K]) decayMax() { i.updateMax(i.max * i.computeExponentialDecayMultiplier()) } -func (i *item) updateMax(score float64) { +func (i *item[K]) updateMax(score float64) { i.max = score i.maxTime = time.Now() } -func (i *item) computeExponentialDecayMultiplier() float64 { +func (i *item[K]) computeExponentialDecayMultiplier() float64 { return math.Pow(0.5, float64(time.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds()) } diff --git a/internal/trending/score.go b/internal/trending/score.go index 22e5f3a..32b5976 100644 --- a/internal/trending/score.go +++ b/internal/trending/score.go @@ -1,7 +1,7 @@ package trending -type score struct { - ID string +type score[K comparable] struct { + ID K Score float64 Probability float64 Expectation float64 @@ -9,28 +9,28 @@ type score struct { KLScore float64 } -type Scores []score +type Scores[K comparable] []score[K] -func (s Scores) Len() int { +func (s Scores[K]) Len() int { return len(s) } -func (s Scores) Swap(i, j int) { +func (s Scores[K]) Swap(i, j int) { s[i], s[j] = s[j], s[i] } -func (s Scores) Less(i, j int) bool { +func (s Scores[K]) Less(i, j int) bool { return s[i].Score > s[j].Score } -func (s Scores) take(count int) Scores { +func (s Scores[K]) take(count int) Scores[K] { if count >= len(s) { return s } return s[0 : count-1] } -func (s Scores) threshold(t float64) Scores { +func (s Scores[K]) threshold(t float64) Scores[K] { for i := range s { if s[i].Score < t { return s[0:i] diff --git a/internal/trending/trending.go b/internal/trending/trending.go index 6117872..136dcba 100644 --- a/internal/trending/trending.go +++ b/internal/trending/trending.go @@ -29,9 +29,9 @@ var defaultBaseCount = 3 var defaultScoreThreshold = 0.01 var defaultCountThreshold = 3.0 -type options struct { - creator TimeSeriesCreator - slidingWindowCreator SlidingWindowCreator +type options[K comparable] struct { + creator TimeSeriesCreator[K] + slidingWindowCreator SlidingWindowCreator[K] halfLife time.Duration @@ -44,59 +44,59 @@ type options struct { countThreshold float64 } -type Option func(*options) +type Option[K comparable] func(*options[K]) -func WithTimeSeries(creator TimeSeriesCreator) Option { - return func(o *options) { +func WithTimeSeries[K comparable](creator TimeSeriesCreator[K]) Option[K] { + return func(o *options[K]) { o.creator = creator } } -func WithSlidingWindow(creator SlidingWindowCreator) Option { - return func(o *options) { +func WithSlidingWindow[K comparable](creator SlidingWindowCreator[K]) Option[K] { + return func(o *options[K]) { o.slidingWindowCreator = creator } } -func WithHalfLife(halfLife time.Duration) Option { - return func(o *options) { +func WithHalfLife[K comparable](halfLife time.Duration) Option[K] { + return func(o *options[K]) { o.halfLife = halfLife } } -func WithRecentDuration(recentDuration time.Duration) Option { - return func(o *options) { +func WithRecentDuration[K comparable](recentDuration time.Duration) Option[K] { + return func(o *options[K]) { o.recentDuration = recentDuration } } -func WithStorageDuration(storageDuration time.Duration) Option { - return func(o *options) { +func WithStorageDuration[K comparable](storageDuration time.Duration) Option[K] { + return func(o *options[K]) { o.storageDuration = storageDuration } } -func WithMaxResults(maxResults int) Option { - return func(o *options) { +func WithMaxResults[K comparable](maxResults int) Option[K] { + return func(o *options[K]) { o.maxResults = maxResults } } -func WithScoreThreshold(threshold float64) Option { - return func(o *options) { +func WithScoreThreshold[K comparable](threshold float64) Option[K] { + return func(o *options[K]) { o.scoreThreshold = threshold } } -func WithCountThreshold(threshold float64) Option { - return func(o *options) { +func WithCountThreshold[K comparable](threshold float64) Option[K] { + return func(o *options[K]) { o.countThreshold = threshold } } -type Scorer struct { - options options - items map[string]*item +type Scorer[K comparable] struct { + options options[K] + items map[K]*item[K] } type SlidingWindow interface { @@ -104,16 +104,16 @@ type SlidingWindow interface { Max() float64 } -type SlidingWindowCreator func(string) SlidingWindow +type SlidingWindowCreator[K comparable] func(K) SlidingWindow type TimeSeries interface { IncreaseAtTime(amount int, time time.Time) Range(start, end time.Time) (float64, error) } -type TimeSeriesCreator func(string) TimeSeries +type TimeSeriesCreator[K comparable] func(K) TimeSeries -func NewMemoryTimeSeries(id string) TimeSeries { +func NewMemoryTimeSeries[K comparable](id K) TimeSeries { ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities( []timeseries.Granularity{ {Granularity: time.Second, Count: 60}, @@ -125,13 +125,13 @@ func NewMemoryTimeSeries(id string) TimeSeries { return ts } -func NewScorer(options ...Option) Scorer { - scorer := Scorer{items: make(map[string]*item)} +func NewScorer[K comparable](options ...Option[K]) Scorer[K] { + scorer := Scorer[K]{items: make(map[K]*item[K])} for _, o := range options { o(&scorer.options) } if scorer.options.creator == nil { - scorer.options.creator = NewMemoryTimeSeries + scorer.options.creator = NewMemoryTimeSeries[K] } if scorer.options.halfLife == 0 { scorer.options.halfLife = defaultHalfLife @@ -155,7 +155,7 @@ func NewScorer(options ...Option) Scorer { scorer.options.baseCount = defaultBaseCount } if scorer.options.slidingWindowCreator == nil { - scorer.options.slidingWindowCreator = func(id string) SlidingWindow { + scorer.options.slidingWindowCreator = func(id K) SlidingWindow { return slidingwindow.NewSlidingWindow( slidingwindow.WithStep(time.Hour*24), slidingwindow.WithDuration(scorer.options.storageDuration), @@ -165,7 +165,7 @@ func NewScorer(options ...Option) Scorer { return scorer } -func (s *Scorer) AddEvent(id string, time time.Time) { +func (s *Scorer[K]) AddEvent(id K, time time.Time) { item := s.items[id] if item == nil { item = newItem(id, &s.options) @@ -174,12 +174,12 @@ func (s *Scorer) AddEvent(id string, time time.Time) { s.addToItem(item, time) } -func (s *Scorer) addToItem(item *item, time time.Time) { +func (s *Scorer[K]) addToItem(item *item[K], time time.Time) { item.eventSeries.IncreaseAtTime(1, time) } -func (s *Scorer) Score() Scores { - var scores Scores +func (s *Scorer[K]) Score() Scores[K] { + var scores Scores[K] for id, item := range s.items { score := item.score() score.ID = id diff --git a/pkg/gordio/server/server.go b/pkg/gordio/server/server.go index 104f31f..90f2fa1 100644 --- a/pkg/gordio/server/server.go +++ b/pkg/gordio/server/server.go @@ -11,6 +11,7 @@ import ( "dynatron.me/x/stillbox/pkg/gordio/database" "dynatron.me/x/stillbox/pkg/gordio/nexus" "dynatron.me/x/stillbox/pkg/gordio/sinks" + "dynatron.me/x/stillbox/pkg/gordio/sinks/alerting" "dynatron.me/x/stillbox/pkg/gordio/sources" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" @@ -43,6 +44,8 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) { return nil, err } + ctx = database.CtxWithDB(ctx, db) + r := chi.NewRouter() authenticator := auth.NewAuthenticator(cfg.Auth) srv := &Server{ @@ -56,6 +59,7 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) { srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true) srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false) + srv.sinks.Register("alerting", alerting.NewSink(ctx), false) srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv)) r.Use(middleware.RequestID) diff --git a/pkg/gordio/sinks/alerting/alerting.go b/pkg/gordio/sinks/alerting/alerting.go new file mode 100644 index 0000000..eaa1e87 --- /dev/null +++ b/pkg/gordio/sinks/alerting/alerting.go @@ -0,0 +1,122 @@ +package alerting + +import ( + "context" + "fmt" + "sync" + "time" + + cl "dynatron.me/x/stillbox/pkg/calls" + "dynatron.me/x/stillbox/pkg/gordio/database" + + "dynatron.me/x/stillbox/internal/timeseries" + "dynatron.me/x/stillbox/internal/trending" + + "github.com/rs/zerolog/log" +) + +const ( + StorageLookbackDays = 2 + HalfLife = time.Hour + RecentDuration = time.Hour +) + +type AlertSink struct { + sync.RWMutex + scorer trending.Scorer[cl.Talkgroup] +} + +func NewSink(ctx context.Context) *AlertSink { + as := &AlertSink{ + scorer: trending.NewScorer[cl.Talkgroup]( + trending.WithTimeSeries(newTimeSeries), + trending.WithStorageDuration[cl.Talkgroup](StorageLookbackDays*24*time.Hour), + trending.WithRecentDuration[cl.Talkgroup](RecentDuration), + trending.WithHalfLife[cl.Talkgroup](HalfLife), + ), + } + + go as.startBackfill(ctx) + + return as +} + +func (as *AlertSink) startBackfill(ctx context.Context) { + since := time.Now().Add(StorageLookbackDays * -24 * time.Hour) + log.Debug().Time("since", since).Msg("starting stats backfill") + count, err := as.backfill(ctx, since) + if err != nil { + log.Error().Err(err).Msg("backfill failed") + return + } + log.Debug().Int("count", count).Int("len", as.scorer.Score().Len()).Msg("backfill finished") + as.printScores() +} + +type score[K comparable] struct { + ID K + Score float64 + Probability float64 + Expectation float64 + Maximum float64 + KLScore float64 +} + +func (as *AlertSink) printScores() { + scores := as.scorer.Score() + fmt.Printf("score len is %d\n", scores.Len()) + for _, s := range scores { + fmt.Printf("%d:%d score %f prob %f exp %f max %f kl %f", s.ID.System, s.ID.Talkgroup, s.Score, + s.Probability, s.Expectation, s.Maximum, s.KLScore) + } +} + +func (as *AlertSink) backfill(ctx context.Context, since time.Time) (count int, err error) { + db := database.FromCtx(ctx) + const backfillStatsQuery = `SELECT system, talkgroup, call_date FROM calls WHERE call_date > $1` + + rows, err := db.Query(ctx, backfillStatsQuery, since) + if err != nil { + return count, err + } + defer rows.Close() + + as.Lock() + defer as.Unlock() + + for rows.Next() { + var tg cl.Talkgroup + var callDate time.Time + if err := rows.Scan(&tg.System, &tg.Talkgroup, &callDate); err != nil { + return count, err + } + as.scorer.AddEvent(tg, callDate) + count++ + } + + if err := rows.Err(); err != nil { + return count, err + } + + return count, nil +} + +func newTimeSeries(id cl.Talkgroup) trending.TimeSeries { + ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities( + []timeseries.Granularity{ + {Granularity: time.Second, Count: 60}, + {Granularity: time.Minute, Count: 10}, + {Granularity: time.Hour, Count: 24}, + {Granularity: time.Hour * 24, Count: StorageLookbackDays}, + }, + )) + return ts +} + +func (as *AlertSink) SinkType() string { + return "alerting" +} + +func (ns *AlertSink) Call(ctx context.Context, call *cl.Call) error { + return nil +}