amigan/stillbox
1
0
Fork 0
Code Issues 16 Pull requests Projects Releases Packages Wiki Activity Actions

Add talkgroup activity alerting #17

Merged
amigan merged 11 commits from alerting into trunk 2024-10-31 00:20:48 -04:00
Conversation 0 Commits 11 Files changed 38 +294 -65
14 changed files with 294 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 55301303ca - Show all commits

8
internal/trending/item.go
View file

@ -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,

4
internal/trending/score.go
View file

@ -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)

6
internal/trending/trending.go
View file

@ -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] {

80
pkg/gordio/sinks/alerting/alerting.go → pkg/gordio/alerting/alerting.go
View file

@ -2,7 +2,7 @@ package alerting
import (
"context"
"fmt"
"sort"
"sync"
"time"
@ -18,26 +18,29 @@ import (
const (
StorageLookbackDays = 7
HalfLife = 30 * time.Minute
RecentDuration = 2*time.Hour
RecentDuration = 2 * time.Hour
ScoreThreshold = -1
CountThreshold = 1
CountThreshold = 1.0
AlerterTickInterval = time.Minute
)
type AlertSink struct {
type Alerter struct {
sync.RWMutex
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)

71
pkg/gordio/alerting/stats.go Normal file
View file

@ -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")
}
}

86
pkg/gordio/alerting/stats.html Normal file
View file

@ -0,0 +1,86 @@
<!DOCTYPE html>
<html>
<head>
<title>Stats</title>
<style>
*, *:before, *:after {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
background: #105469;
font-family: sans-serif;
}
table {
background: #012B39;
border-radius: 0.25em;
border-collapse: collapse;
margin-top: 1em;
margin-bottom: 1em;
margin-left: auto;
margin-right: auto;
}
th {
border-bottom: 1px solid #364043;
color: #E2B842;
font-size: 0.85em;
font-weight: 600;
padding: 0.5em 1em;
text-align: left;
}
td {
color: #fff;
font-weight: 400;
padding: 0.65em 1em;
}
.disabled td {
color: #4F5F64;
}
tbody tr {
transition: background 0.25s ease;
}
tbody tr:hover {
background: #014055;
}
</style>
</head>
<body>
<table>
<tr>
<th>System</th>
<th>TG</th>
<th>TG ID</th>
<th>Count</th>
<th>Recent</th>
<th>Score</th>
<th>Probab</th>
<th>Expect</th>
<th>Max</th>
<th>KL</th>
</tr>
{{ range .Scores }}
{{ $tg := (index $.TGs .ID) }}
<tr>
<td>{{ $tg.Name_2}}</td>
<td>{{ $tg.Name}}</td>
<td>{{ .ID.Talkgroup }}</td>
<td>{{ .Count }}</td>
<td>{{ f .RecentCount }}</td>
<td>{{ f .Score }}</td>
<td>{{ f .Probability }}</td>
<td>{{ f .Expectation }}</td>
<td>{{ f .Maximum }}</td>
<td>{{ f .KLScore }}</td>
</tr>
{{else}}
<tr>
<td colspan="10">No Data</td>
</tr>
{{ end }}
<tr>
<td colspan="10">Last updated at {{ .LastScore }}</td>
</tr>
</table>
</body>
</html>

2
pkg/gordio/database/database.go
View file

@ -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

1
pkg/gordio/database/querier.go
View file

@ -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)

52
pkg/gordio/database/talkgroups.sql.go
View file

@ -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]

1
pkg/gordio/server/routes.go
View file

@ -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) {

9
pkg/gordio/server/server.go
View file

@ -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
}
@ -55,11 +56,12 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
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()

9
pkg/gordio/sinks/alerter.go Normal file
View file

@ -0,0 +1,9 @@
package sinks
import (
"dynatron.me/x/stillbox/pkg/gordio/alerting"
)
func NewAlerterSink(a *alerting.Alerter) *alerting.Alerter {
return a
}

7
pkg/gordio/sinks/sinks.go
View file

@ -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()

5
sql/postgres/queries/talkgroups.sql
View file

@ -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,