From 55301303ca30f3af7b9ba0b042c33eae32097af7 Mon Sep 17 00:00:00 2001 From: Daniel Ponte Date: Thu, 24 Oct 2024 20:00:46 -0400 Subject: [PATCH] wip --- internal/trending/item.go | 8 +- internal/trending/score.go | 4 +- internal/trending/trending.go | 6 +- pkg/gordio/{sinks => }/alerting/alerting.go | 86 ++++++++++----------- pkg/gordio/alerting/stats.go | 71 +++++++++++++++++ pkg/gordio/alerting/stats.html | 86 +++++++++++++++++++++ pkg/gordio/database/database.go | 2 - pkg/gordio/database/querier.go | 1 + pkg/gordio/database/talkgroups.sql.go | 52 +++++++++++++ pkg/gordio/server/routes.go | 1 + pkg/gordio/server/server.go | 21 +++-- pkg/gordio/sinks/alerter.go | 9 +++ pkg/gordio/sinks/sinks.go | 7 ++ sql/postgres/queries/talkgroups.sql | 5 ++ 14 files changed, 294 insertions(+), 65 deletions(-) rename pkg/gordio/{sinks => }/alerting/alerting.go (61%) create mode 100644 pkg/gordio/alerting/stats.go create mode 100644 pkg/gordio/alerting/stats.html create mode 100644 pkg/gordio/sinks/alerter.go diff --git a/internal/trending/item.go b/internal/trending/item.go index 665bbf1..625a943 100644 --- a/internal/trending/item.go +++ b/internal/trending/item.go @@ -33,10 +33,10 @@ func newItem[K comparable](id K, options *options[K]) *item[K] { } } -func (i *item[K]) score() score[K] { +func (i *item[K]) score() Score[K] { recentCount, count := i.computeCounts() if recentCount < i.options.countThreshold { - return score[K]{} + return Score[K]{} } if recentCount == count { // we see this for the first time so there is no historical data @@ -60,9 +60,9 @@ func (i *item[K]) score() score[K] { } i.decayMax() - mixedScore := 0.5 * (klScore + i.max) + mixedScore := 5 * (klScore + i.max) - return score[K]{ + return Score[K]{ Score: mixedScore, Probability: probability, Expectation: expectation, diff --git a/internal/trending/score.go b/internal/trending/score.go index dbbc4ca..f59368b 100644 --- a/internal/trending/score.go +++ b/internal/trending/score.go @@ -1,6 +1,6 @@ package trending -type score[K comparable] struct { +type Score[K comparable] struct { ID K Score float64 Probability float64 @@ -11,7 +11,7 @@ type score[K comparable] struct { RecentCount float64 } -type Scores[K comparable] []score[K] +type Scores[K comparable] []Score[K] func (s Scores[K]) Len() int { return len(s) diff --git a/internal/trending/trending.go b/internal/trending/trending.go index e25c2c6..20e1df2 100644 --- a/internal/trending/trending.go +++ b/internal/trending/trending.go @@ -1,7 +1,6 @@ package trending import ( - "fmt" "sort" "time" @@ -175,9 +174,8 @@ func (s *Scorer[K]) AddEvent(id K, time time.Time) { s.addToItem(item, time) } -func (s *Scorer[K]) addToItem(item *item[K], time time.Time) { - fmt.Println("add", time.String()) - item.eventSeries.IncreaseAtTime(1, time) +func (s *Scorer[K]) addToItem(item *item[K], tm time.Time) { + item.eventSeries.IncreaseAtTime(1, tm) } func (s *Scorer[K]) Score() Scores[K] { diff --git a/pkg/gordio/sinks/alerting/alerting.go b/pkg/gordio/alerting/alerting.go similarity index 61% rename from pkg/gordio/sinks/alerting/alerting.go rename to pkg/gordio/alerting/alerting.go index a29bf09..cb5192a 100644 --- a/pkg/gordio/sinks/alerting/alerting.go +++ b/pkg/gordio/alerting/alerting.go @@ -2,7 +2,7 @@ package alerting import ( "context" - "fmt" + "sort" "sync" "time" @@ -16,28 +16,31 @@ import ( ) const ( - StorageLookbackDays = 7 + StorageLookbackDays = 7 HalfLife = 30 * time.Minute - RecentDuration = 2*time.Hour - ScoreThreshold = -1 - CountThreshold = 1 + RecentDuration = 2 * time.Hour + ScoreThreshold = -1 + CountThreshold = 1.0 + AlerterTickInterval = time.Minute ) -type AlertSink struct { +type Alerter struct { sync.RWMutex - scorer trending.Scorer[cl.Talkgroup] + scorer trending.Scorer[cl.Talkgroup] + scores trending.Scores[cl.Talkgroup] + lastScore time.Time } type myClock struct { offset time.Duration - } + func (c *myClock) Now() time.Time { return time.Now().Add(c.offset) } -func NewSink(ctx context.Context) *AlertSink { - as := &AlertSink{ +func New() *Alerter { + as := &Alerter{ scorer: trending.NewScorer[cl.Talkgroup]( trending.WithTimeSeries(newTimeSeries), trending.WithStorageDuration[cl.Talkgroup](StorageLookbackDays*24*time.Hour), @@ -48,11 +51,27 @@ func NewSink(ctx context.Context) *AlertSink { ), } - go as.startBackfill(ctx) - return as } +func (as *Alerter) Go(ctx context.Context) { + as.startBackfill(ctx) + + as.score(ctx, time.Now()) + ticker := time.NewTicker(AlerterTickInterval) + + for { + select { + case now := <-ticker.C: + as.score(ctx, now) + case <-ctx.Done(): + ticker.Stop() + return + } + } + +} + func newTimeSeries(id cl.Talkgroup) trending.TimeSeries { ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities( []timeseries.Granularity{ @@ -65,9 +84,8 @@ func newTimeSeries(id cl.Talkgroup) trending.TimeSeries { return ts } -func (as *AlertSink) startBackfill(ctx context.Context) { +func (as *Alerter) startBackfill(ctx context.Context) { now := time.Now() - cl := &myClock{-24*StorageLookbackDays*time.Hour} since := now.Add(StorageLookbackDays * -24 * time.Hour) log.Debug().Time("since", since).Msg("starting stats backfill") count, err := as.backfill(ctx, since) @@ -76,42 +94,20 @@ func (as *AlertSink) startBackfill(ctx context.Context) { return } log.Debug().Int("count", count).Str("in", time.Now().Sub(now).String()).Int("len", as.scorer.Score().Len()).Msg("backfill finished") - timeseries.DefaultClock = cl - for { - fmt.Printf("offs: %s (%s)\n", cl.offset.String(), cl.Now().String()) - as.printScores(ctx) - cl.offset += time.Minute*5 - if cl.offset == time.Minute*5 { - break - } - } } -func (as *AlertSink) printScores(ctx context.Context) { - db := database.FromCtx(ctx) +func (as *Alerter) score(ctx context.Context, now time.Time) { as.Lock() defer as.Unlock() - scores := as.scorer.Score() - //fmt.Printf("score len is %d\n", scores.Len()) - //const scoreMult = 1000000000 - const scoreMult = 1 - for _, s := range scores { - if s.ID.Talkgroup != 1185 { - continue - } - tg, _ := db.GetTalkgroup(ctx, int(s.ID.System), int(s.ID.Talkgroup)) - tgn := "" - if tg.Name != nil { - tgn = *tg.Name - } - fmt.Printf("%s\t\t\t%d:%d c %f\trc %f\tscore %f\tprob %f\texp %f\tmax %f\tkl %f\n", tgn, s.ID.System, s.ID.Talkgroup, - s.Count, s.RecentCount, s.Score*scoreMult, s.Probability, s.Expectation, s.Maximum, s.KLScore) - } + + as.scores = as.scorer.Score() + as.lastScore = now + sort.Sort(as.scores) } -func (as *AlertSink) backfill(ctx context.Context, since time.Time) (count int, err error) { +func (as *Alerter) 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 AND call_date < $2 AND talkgroup = 1185 ` + const backfillStatsQuery = `SELECT system, talkgroup, call_date FROM calls WHERE call_date > $1 AND call_date < $2` rows, err := db.Query(ctx, backfillStatsQuery, since, timeseries.DefaultClock.Now()) if err != nil { @@ -139,11 +135,11 @@ func (as *AlertSink) backfill(ctx context.Context, since time.Time) (count int, return count, nil } -func (as *AlertSink) SinkType() string { +func (as *Alerter) SinkType() string { return "alerting" } -func (as *AlertSink) Call(ctx context.Context, call *cl.Call) error { +func (as *Alerter) Call(ctx context.Context, call *cl.Call) error { as.Lock() defer as.Unlock() as.scorer.AddEvent(call.TalkgroupTuple(), call.DateTime) diff --git a/pkg/gordio/alerting/stats.go b/pkg/gordio/alerting/stats.go new file mode 100644 index 0000000..9e1fafa --- /dev/null +++ b/pkg/gordio/alerting/stats.go @@ -0,0 +1,71 @@ +package alerting + +import ( + _ "embed" + "fmt" + "html/template" + "net/http" + "time" + + "dynatron.me/x/stillbox/pkg/calls" + "dynatron.me/x/stillbox/pkg/gordio/database" + + "dynatron.me/x/stillbox/internal/trending" + + "github.com/go-chi/chi/v5" + "github.com/rs/zerolog/log" +) + +//go:embed stats.html +var statsTemplateFile string + +var ( + funcMap = template.FuncMap{ + "f": func(v float64) string { + return fmt.Sprintf("%.4f", v) + }, + } + statTmpl = template.Must(template.New("stats").Funcs(funcMap).Parse(statsTemplateFile)) +) + +func (as *Alerter) PrivateRoutes(r chi.Router) { + r.Get("/tgstats", as.tgStats) +} + +func (as *Alerter) tgStats(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + db := database.FromCtx(ctx) + + packed := make([]int64, 0, len(as.scores)) + for _, s := range as.scores { + packed = append(packed, s.ID.Pack()) + } + + tgs, err := db.GetTalkgroupsByPackedIDs(ctx, packed) + if err != nil { + log.Error().Err(err).Msg("stats TG get failed") + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + tgMap := make(map[calls.Talkgroup]database.GetTalkgroupsByPackedIDsRow, len(tgs)) + for _, t := range tgs { + tgMap[calls.Talkgroup{System: uint32(t.SystemID), Talkgroup: uint32(t.ID)}] = t + } + + renderData := struct { + TGs map[calls.Talkgroup]database.GetTalkgroupsByPackedIDsRow + Scores trending.Scores[calls.Talkgroup] + LastScore time.Time + }{ + TGs: tgMap, + Scores: as.scores, + LastScore: as.lastScore, + } + + w.WriteHeader(http.StatusOK) + err = statTmpl.Execute(w, renderData) + if err != nil { + log.Error().Err(err).Msg("stat template exec") + } +} diff --git a/pkg/gordio/alerting/stats.html b/pkg/gordio/alerting/stats.html new file mode 100644 index 0000000..61e4e56 --- /dev/null +++ b/pkg/gordio/alerting/stats.html @@ -0,0 +1,86 @@ + + + + Stats + + + + + + + + + + + + + + + + + {{ range .Scores }} + {{ $tg := (index $.TGs .ID) }} + + + + + + + + + + + + + {{else}} + + + + {{ end }} + + + +
SystemTGTG IDCountRecentScoreProbabExpectMaxKL
{{ $tg.Name_2}}{{ $tg.Name}}{{ .ID.Talkgroup }}{{ .Count }}{{ f .RecentCount }}{{ f .Score }}{{ f .Probability }}{{ f .Expectation }}{{ f .Maximum }}{{ f .KLScore }}
No Data
Last updated at {{ .LastScore }}
+ + diff --git a/pkg/gordio/database/database.go b/pkg/gordio/database/database.go index d2590c1..591f02c 100644 --- a/pkg/gordio/database/database.go +++ b/pkg/gordio/database/database.go @@ -13,8 +13,6 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -// This file will eventually turn into a postgres driver. - // DB is a database handle. type DB struct { *pgxpool.Pool diff --git a/pkg/gordio/database/querier.go b/pkg/gordio/database/querier.go index 0f9d2a4..50ffb6b 100644 --- a/pkg/gordio/database/querier.go +++ b/pkg/gordio/database/querier.go @@ -24,6 +24,7 @@ type Querier interface { GetTalkgroupIDsByTags(ctx context.Context, anytags []string, alltags []string, nottags []string) ([]GetTalkgroupIDsByTagsRow, error) GetTalkgroupTags(ctx context.Context, sys int, tg int) ([]string, error) GetTalkgroupWithLearned(ctx context.Context, systemID int, tgid int) (GetTalkgroupWithLearnedRow, error) + GetTalkgroupsByPackedIDs(ctx context.Context, dollar_1 []int64) ([]GetTalkgroupsByPackedIDsRow, error) GetTalkgroupsWithAllTags(ctx context.Context, tags []string) ([]Talkgroup, error) GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) ([]Talkgroup, error) GetUserByID(ctx context.Context, id int32) (User, error) diff --git a/pkg/gordio/database/talkgroups.sql.go b/pkg/gordio/database/talkgroups.sql.go index 98d7788..6ff6b1e 100644 --- a/pkg/gordio/database/talkgroups.sql.go +++ b/pkg/gordio/database/talkgroups.sql.go @@ -137,6 +137,58 @@ func (q *Queries) GetTalkgroupWithLearned(ctx context.Context, systemID int, tgi return i, err } +const getTalkgroupsByPackedIDs = `-- name: GetTalkgroupsByPackedIDs :many +SELECT tg.id, system_id, tgid, tg.name, alpha_tag, tg_group, frequency, metadata, tags, sys.id, sys.name FROM talkgroups tg +JOIN systems sys ON tg.system_id = sys.id +WHERE tg.id = ANY($1::INT8[]) +` + +type GetTalkgroupsByPackedIDsRow struct { + ID int64 `json:"id"` + SystemID int32 `json:"system_id"` + Tgid int32 `json:"tgid"` + Name *string `json:"name"` + AlphaTag *string `json:"alpha_tag"` + TgGroup *string `json:"tg_group"` + Frequency *int32 `json:"frequency"` + Metadata []byte `json:"metadata"` + Tags []string `json:"tags"` + ID_2 int `json:"id_2"` + Name_2 string `json:"name_2"` +} + +func (q *Queries) GetTalkgroupsByPackedIDs(ctx context.Context, dollar_1 []int64) ([]GetTalkgroupsByPackedIDsRow, error) { + rows, err := q.db.Query(ctx, getTalkgroupsByPackedIDs, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetTalkgroupsByPackedIDsRow + for rows.Next() { + var i GetTalkgroupsByPackedIDsRow + if err := rows.Scan( + &i.ID, + &i.SystemID, + &i.Tgid, + &i.Name, + &i.AlphaTag, + &i.TgGroup, + &i.Frequency, + &i.Metadata, + &i.Tags, + &i.ID_2, + &i.Name_2, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getTalkgroupsWithAllTags = `-- name: GetTalkgroupsWithAllTags :many SELECT id, system_id, tgid, name, alpha_tag, tg_group, frequency, metadata, tags FROM talkgroups WHERE tags && ARRAY[$1] diff --git a/pkg/gordio/server/routes.go b/pkg/gordio/server/routes.go index 1ba9d77..601c2db 100644 --- a/pkg/gordio/server/routes.go +++ b/pkg/gordio/server/routes.go @@ -33,6 +33,7 @@ func (s *Server) setupRoutes() { r.Use(s.auth.VerifyMiddleware(), s.auth.AuthMiddleware()) s.nex.PrivateRoutes(r) s.auth.PrivateRoutes(r) + s.alerter.PrivateRoutes(r) }) r.Group(func(r chi.Router) { diff --git a/pkg/gordio/server/server.go b/pkg/gordio/server/server.go index 90f2fa1..bd64082 100644 --- a/pkg/gordio/server/server.go +++ b/pkg/gordio/server/server.go @@ -6,12 +6,12 @@ import ( "os" "time" + "dynatron.me/x/stillbox/pkg/gordio/alerting" "dynatron.me/x/stillbox/pkg/gordio/auth" "dynatron.me/x/stillbox/pkg/gordio/config" "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" @@ -30,6 +30,7 @@ type Server struct { sinks sinks.Sinks nex *nexus.Nexus logger *Logger + alerter *alerting.Alerter hup chan os.Signal } @@ -49,17 +50,18 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) { r := chi.NewRouter() authenticator := auth.NewAuthenticator(cfg.Auth) srv := &Server{ - auth: authenticator, - conf: cfg, - db: db, - r: r, - nex: nexus.New(), - logger: logger, + auth: authenticator, + conf: cfg, + db: db, + r: r, + nex: nexus.New(), + logger: logger, + alerter: alerting.New(), } 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.sinks.Register("alerting", sinks.NewAlerterSink(srv.alerter), false) srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv)) r.Use(middleware.RequestID) @@ -92,6 +94,7 @@ func (s *Server) Go(ctx context.Context) error { } go s.nex.Go(ctx) + go s.alerter.Go(ctx) var err error go func() { @@ -99,6 +102,8 @@ func (s *Server) Go(ctx context.Context) error { }() <-ctx.Done() + s.sinks.Shutdown() + ctxShutdown, cancel := context.WithTimeout(context.Background(), shutdownTimeout) defer cancel() diff --git a/pkg/gordio/sinks/alerter.go b/pkg/gordio/sinks/alerter.go new file mode 100644 index 0000000..1b63bd8 --- /dev/null +++ b/pkg/gordio/sinks/alerter.go @@ -0,0 +1,9 @@ +package sinks + +import ( + "dynatron.me/x/stillbox/pkg/gordio/alerting" +) + +func NewAlerterSink(a *alerting.Alerter) *alerting.Alerter { + return a +} diff --git a/pkg/gordio/sinks/sinks.go b/pkg/gordio/sinks/sinks.go index 7495370..3c8afd9 100644 --- a/pkg/gordio/sinks/sinks.go +++ b/pkg/gordio/sinks/sinks.go @@ -40,6 +40,13 @@ func (s *Sinks) Register(name string, toAdd Sink, required bool) { }) } +func (s *Sinks) Shutdown() { + s.Lock() + defer s.Unlock() + + s.sinks = nil +} + func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) error { s.Lock() defer s.Unlock() diff --git a/sql/postgres/queries/talkgroups.sql b/sql/postgres/queries/talkgroups.sql index e316dae..fe902e9 100644 --- a/sql/postgres/queries/talkgroups.sql +++ b/sql/postgres/queries/talkgroups.sql @@ -28,6 +28,11 @@ WHERE id = ANY($1); SELECT * FROM talkgroups WHERE id = systg2id(sqlc.arg(system_id), sqlc.arg(tgid)); +-- name: GetTalkgroupsByPackedIDs :many +SELECT * FROM talkgroups tg +JOIN systems sys ON tg.system_id = sys.id +WHERE tg.id = ANY($1::INT8[]); + -- name: GetTalkgroupWithLearned :one SELECT tg.id, tg.system_id, sys.name system_name, tg.tgid, tg.name,