mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-22 17:41:26 +00:00
feat(stats): add bounded path and hash-width analytics (#159)
This commit is contained in:
n30nex authored and GitHub committed
1 parent
6107578c25
commit
8a3fa7e6bf
25 files changed
+1205
-4
No files matched your search
@@ -0,0 +1,31 @@
|
||||
-- Classify once per background refresh, preserving the decoder's header rules.
|
||||
-- No observation rows or deduplication constraints are changed.
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS mv_path_stats_hourly AS
|
||||
WITH classified AS (
|
||||
SELECT iata, heard_at, hash_size, hop_count,
|
||||
CASE WHEN payload_type = 9 THEN 2
|
||||
-- Match meshcore-go IsValidPathLen (1/2/3-byte hashes, max 64 path bytes).
|
||||
WHEN payload_type IS NULL OR payload_type NOT BETWEEN 0 AND 15
|
||||
OR NOT (path_length_byte BETWEEN 0 AND 191
|
||||
AND hash_size BETWEEN 1 AND 3 AND hop_count BETWEEN 0 AND 63
|
||||
AND hash_size = (path_length_byte >> 6) + 1
|
||||
AND hop_count = (path_length_byte & 63)
|
||||
AND hash_size::integer * hop_count::integer <= 64
|
||||
AND COALESCE(octet_length(path_bytes), 0) = hash_size::integer * hop_count::integer)
|
||||
THEN 3
|
||||
WHEN hop_count = 0 THEN 1
|
||||
ELSE 0 END::integer AS category
|
||||
FROM packet_observations
|
||||
WHERE heard_at >= date_trunc('hour', NOW(), 'UTC') - INTERVAL '720 hours'
|
||||
AND heard_at < date_trunc('hour', NOW(), 'UTC')
|
||||
), buckets AS (
|
||||
SELECT iata, date_trunc('hour', heard_at, 'UTC') AS hour, category,
|
||||
CASE WHEN category = 0 THEN hash_size ELSE 0 END::integer AS hash_bytes,
|
||||
CASE WHEN category = 0 THEN hop_count ELSE 0 END::integer AS entries
|
||||
FROM classified
|
||||
)
|
||||
SELECT iata, hour, category, hash_bytes, entries, count(*)::bigint AS receptions
|
||||
FROM buckets GROUP BY iata, hour, category, hash_bytes, entries;
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_mv_path_stats_hourly
|
||||
ON mv_path_stats_hourly (iata, hour, category, hash_bytes, entries);
|
||||
CREATE INDEX IF NOT EXISTS idx_mv_path_stats_hourly_hour ON mv_path_stats_hourly (hour);
|
||||
+76
@@ -0,0 +1,76 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
func (s *Store) GetPathStats(ctx context.Context, since, until time.Time, iatas []string) (*api.PathStats, error) {
|
||||
rows, err := s.q.GetPathStats(ctx, sqlc.GetPathStatsParams{Since: pgtype.Timestamptz{Time: since, Valid: true}, Until: pgtype.Timestamptz{Time: until, Valid: true}, Iatas: iatas})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stats := &api.PathStats{Since: since.UnixMilli(), Until: until.UnixMilli(), HashWidths: []api.PathHashWidth{{Bytes: 1}, {Bytes: 2}, {Bytes: 3}}, PathLengths: []api.PathLengthBin{}, Hourly: []api.PathHour{}}
|
||||
var lengths [64]int64
|
||||
for _, row := range rows {
|
||||
if row.Category == 0 && (row.HashBytes < 1 || int(row.HashBytes) > len(stats.HashWidths) || row.Entries < 0 || int(row.Entries) >= len(lengths)) {
|
||||
return nil, fmt.Errorf("invalid path statistics bucket")
|
||||
}
|
||||
if row.IsHourly {
|
||||
hour := row.Hour.Time.UnixMilli()
|
||||
if len(stats.Hourly) == 0 || stats.Hourly[len(stats.Hourly)-1].Hour != hour {
|
||||
stats.Hourly = append(stats.Hourly, api.PathHour{Hour: hour})
|
||||
}
|
||||
point := &stats.Hourly[len(stats.Hourly)-1]
|
||||
point.Receptions += row.Receptions
|
||||
switch row.Category {
|
||||
case 0:
|
||||
switch row.HashBytes {
|
||||
case 1:
|
||||
point.OneByte += row.Receptions
|
||||
case 2:
|
||||
point.TwoByte += row.Receptions
|
||||
case 3:
|
||||
point.ThreeByte += row.Receptions
|
||||
}
|
||||
case 1:
|
||||
point.Empty += row.Receptions
|
||||
case 2:
|
||||
point.Trace += row.Receptions
|
||||
case 3:
|
||||
point.Unclassified += row.Receptions
|
||||
}
|
||||
continue
|
||||
}
|
||||
stats.Receptions += row.Receptions
|
||||
switch row.Category {
|
||||
case 0:
|
||||
stats.Hashed += row.Receptions
|
||||
stats.HashWidths[row.HashBytes-1].Receptions += row.Receptions
|
||||
lengths[row.Entries] += row.Receptions
|
||||
case 1:
|
||||
stats.Empty += row.Receptions
|
||||
lengths[0] += row.Receptions
|
||||
case 2:
|
||||
stats.Trace += row.Receptions
|
||||
case 3:
|
||||
stats.Unclassified += row.Receptions
|
||||
}
|
||||
}
|
||||
for entries, count := range lengths {
|
||||
if count > 0 {
|
||||
stats.PathLengths = append(stats.PathLengths, api.PathLengthBin{Entries: int32(entries), Receptions: count})
|
||||
}
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
func (s *Store) RefreshPathStats(ctx context.Context) error { return s.q.RefreshPathStats(ctx) }
|
||||
@@ -0,0 +1,163 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api"
|
||||
"github.com/MeshCore-Beacon/beacon-server/internal/api/handlers"
|
||||
"github.com/jackc/pgx/v5"
|
||||
meshcore "github.com/meshcore-go/meshcore-go"
|
||||
)
|
||||
|
||||
func TestPathsPostgres(t *testing.T) {
|
||||
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
conn, err := pgx.Connect(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close(context.Background())
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
isolateStatsSchema(t, ctx, tx)
|
||||
_, err = tx.Exec(ctx, `
|
||||
SET LOCAL TIME ZONE 'America/Vancouver';
|
||||
CREATE TABLE packet_observations(heard_at timestamptz NOT NULL,iata char(3) NOT NULL,payload_type smallint,path_length_byte smallint NOT NULL,hash_size smallint NOT NULL,hop_count smallint NOT NULL,path_bytes bytea);
|
||||
INSERT INTO packet_observations
|
||||
SELECT CASE id WHEN 1 THEN '2026-01-01 00:30+00'::timestamptz WHEN 17 THEN '2026-01-01 04:00+00'::timestamptz WHEN 18 THEN '2026-01-01 04:30+00'::timestamptz WHEN 20 THEN '2026-01-01 03:00+00'::timestamptz
|
||||
ELSE '2026-01-01 00:30+00'::timestamptz+(id/10)*interval '2 hour'+(id%10)*interval '1 minute' END,
|
||||
iata,payload,raw,width,hops,CASE WHEN size IS NULL THEN NULL ELSE decode(repeat('ab',size),'hex') END
|
||||
FROM (VALUES
|
||||
(1,'YVR',4,2,1,2,2),(2,'YVR',4,67,2,3,6),(3,'YYJ',5,130,3,2,6),
|
||||
(4,'YVR',4,64,2,0,NULL),(5,'YVR',9,3,1,3,3),(6,'YYJ',NULL,2,1,2,2),
|
||||
(7,'YVR',4,3,1,3,2),(8,'YVR',4,194,4,2,8),(9,'YVR',4,96,2,32,64),
|
||||
(10,'YVR',4,97,2,33,66),(11,'YVR',4,64,1,0,0),(12,'YVR',4,0,1,1,1),
|
||||
(13,'YYZ',5,0,1,0,0),(14,'YVR',16,1,1,1,1),(15,'YVR',4,63,1,63,63),
|
||||
(16,'YVR',4,149,3,21,63),(17,'YVR',4,1,1,1,1),(18,'YVR',4,1,1,1,1),
|
||||
(19,'YYJ',9,3,1,3,3),(20,'YVR',4,64,2,0,0)
|
||||
) v(id,iata,payload,raw,width,hops,size);`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
since := time.Now().UTC().Truncate(24 * time.Hour).Add(-48 * time.Hour)
|
||||
if _, err := tx.Exec(ctx, "UPDATE packet_observations SET heard_at=to_timestamp(extract(epoch FROM $1::timestamptz)+extract(epoch FROM heard_at)-extract(epoch FROM '2026-01-01 00:00+00'::timestamptz))", since); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
applyStatsMigration(t, ctx, tx, "036_mv_path_stats.sql")
|
||||
applyStatsMigration(t, ctx, tx, "036_mv_path_stats.sql") // interrupted journal retry
|
||||
store := &Store{q: sqlc.New(tx)}
|
||||
until := since.Add(4 * time.Hour)
|
||||
for _, mode := range []string{"force_custom_plan", "force_generic_plan"} {
|
||||
if _, err := tx.Exec(ctx, "SET LOCAL plan_cache_mode="+mode); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
iatas []string
|
||||
counts [5]int64
|
||||
}{
|
||||
{"global", nil, [5]int64{18, 6, 3, 2, 7}}, {"empty filter", []string{}, [5]int64{18, 6, 3, 2, 7}},
|
||||
{"YVR", []string{"YVR"}, [5]int64{14, 5, 2, 1, 6}}, {"overlap", []string{"YVR", "YYJ", "YVR"}, [5]int64{17, 6, 2, 2, 7}},
|
||||
{"unknown", []string{"ZZZ"}, [5]int64{}}, {"empty region", []string{""}, [5]int64{}},
|
||||
} {
|
||||
t.Run(mode+"/"+tc.name, func(t *testing.T) {
|
||||
got, err := store.GetPathStats(ctx, since, until, tc.iatas)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if [5]int64{got.Receptions, got.Hashed, got.Empty, got.Trace, got.Unclassified} != tc.counts {
|
||||
t.Fatalf("counts: %+v", got)
|
||||
}
|
||||
var widths, lengths, hours int64
|
||||
for _, b := range got.HashWidths {
|
||||
widths += b.Receptions
|
||||
}
|
||||
for _, b := range got.PathLengths {
|
||||
lengths += b.Receptions
|
||||
}
|
||||
for i, h := range got.Hourly {
|
||||
hours += h.Receptions
|
||||
if h.Receptions != h.OneByte+h.TwoByte+h.ThreeByte+h.Empty+h.Trace+h.Unclassified || h.Hour%3600000 != 0 || (i > 0 && h.Hour <= got.Hourly[i-1].Hour) {
|
||||
t.Fatalf("hour: %+v", h)
|
||||
}
|
||||
}
|
||||
if widths != got.Hashed || lengths != got.Hashed+got.Empty || hours != got.Receptions {
|
||||
t.Fatal("partitions do not reconcile")
|
||||
}
|
||||
if tc.name == "global" && (got.HashWidths[0].Receptions != 2 || got.HashWidths[1].Receptions != 2 || got.HashWidths[2].Receptions != 2 || got.PathLengths[0].Entries != 0 || got.PathLengths[0].Receptions != 3 || len(got.Hourly) != 3) {
|
||||
t.Fatalf("empty/trace widths or missing hour: %+v", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
handlers.StatsRouter(store).ServeHTTP(w, httptest.NewRequest("GET", fmt.Sprintf("/paths?since=%d&until=%d&iatas=YVR", since.UnixMilli()+123, until.UnixMilli()+123), nil).WithContext(ctx))
|
||||
var response api.PathStats
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || w.Code != 200 || response.Receptions != 14 || response.Hashed != 5 {
|
||||
t.Fatalf("HTTP %d: %s (%v)", w.Code, w.Body.String(), err)
|
||||
}
|
||||
if _, err = tx.Exec(ctx, "TRUNCATE packet_observations"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `INSERT INTO packet_observations SELECT $1::timestamptz+interval '1 hour','YVR',4,i,(i>>6)+1,i&63,decode(repeat('ab',((i>>6)+1)*(i&63)),'hex') FROM generate_series(0,255)i;`, since); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stale, err := store.GetPathStats(ctx, since, until, nil)
|
||||
if err != nil || stale.Receptions != 18 {
|
||||
t.Fatalf("snapshot lost: %+v %v", stale, err)
|
||||
}
|
||||
if err := store.RefreshPathStats(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := store.GetPathStats(ctx, since, until, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var hashed, empty, unknown int64
|
||||
var widths [3]int64
|
||||
for raw := 0; raw < 256; raw++ {
|
||||
if !meshcore.IsValidPathLen(uint8(raw)) {
|
||||
unknown++
|
||||
} else if raw&63 == 0 {
|
||||
empty++
|
||||
} else {
|
||||
hashed++
|
||||
widths[raw>>6]++
|
||||
}
|
||||
}
|
||||
if got.Hashed != hashed || got.Empty != empty || got.Unclassified != unknown || got.Receptions != 256 {
|
||||
t.Fatalf("decoder agreement: %+v", got)
|
||||
}
|
||||
for i, count := range widths {
|
||||
if got.HashWidths[i].Receptions != count {
|
||||
t.Fatal("decoder width disagreement")
|
||||
}
|
||||
}
|
||||
if _, err = tx.Exec(ctx, "TRUNCATE packet_observations"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.RefreshPathStats(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err = store.GetPathStats(ctx, since, until, nil)
|
||||
if err != nil || got.Receptions != 0 || got.Hourly == nil || got.PathLengths == nil || len(got.HashWidths) != 3 {
|
||||
t.Fatalf("empty: %+v %v", got, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
// Copyright 2026 Beacon Contributors
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
|
||||
)
|
||||
|
||||
type pathBucketQuerier struct {
|
||||
sqlc.Querier
|
||||
row sqlc.GetPathStatsRow
|
||||
}
|
||||
|
||||
func (q pathBucketQuerier) GetPathStats(context.Context, sqlc.GetPathStatsParams) ([]sqlc.GetPathStatsRow, error) {
|
||||
return []sqlc.GetPathStatsRow{q.row}, nil
|
||||
}
|
||||
|
||||
func TestPathStatsRejectsInvalidDatabaseBucket(t *testing.T) {
|
||||
for _, row := range []sqlc.GetPathStatsRow{
|
||||
{HashBytes: 0}, {HashBytes: 4}, {HashBytes: 1, Entries: -1}, {HashBytes: 1, Entries: 64},
|
||||
{HashBytes: 4, IsHourly: true},
|
||||
} {
|
||||
store := &Store{q: pathBucketQuerier{row: row}}
|
||||
if _, err := store.GetPathStats(context.Background(), time.Time{}, time.Time{}, nil); err == nil {
|
||||
t.Fatalf("invalid bucket accepted: %+v", row)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
-- name: GetPathStats :many
|
||||
-- All HTTP aggregates read the compact classification snapshot.
|
||||
WITH readings AS (
|
||||
SELECT * FROM mv_path_stats_hourly
|
||||
WHERE hour >= @since::timestamptz AND hour < @until::timestamptz
|
||||
AND COALESCE(cardinality(@iatas::bpchar[]), 0) = 0
|
||||
UNION ALL
|
||||
SELECT * FROM mv_path_stats_hourly
|
||||
WHERE hour >= @since::timestamptz AND hour < @until::timestamptz
|
||||
AND cardinality(@iatas::bpchar[]) > 0 AND iata = ANY(@iatas::bpchar[])
|
||||
)
|
||||
SELECT (grouping(hour) = 0)::boolean AS is_hourly,
|
||||
COALESCE(hour, 'epoch'::timestamptz)::timestamptz AS hour,
|
||||
category, hash_bytes, COALESCE(entries, 0)::integer AS entries,
|
||||
sum(receptions)::bigint AS receptions
|
||||
FROM readings
|
||||
GROUP BY GROUPING SETS ((category, hash_bytes, entries), (hour, category, hash_bytes))
|
||||
ORDER BY is_hourly, hour, category, hash_bytes, entries;
|
||||
|
||||
-- name: RefreshPathStats :exec
|
||||
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_path_stats_hourly;
|
||||
@@ -577,6 +577,21 @@ func (mr *MockQuerierMockRecorder) GetPacketsByTraceTag(ctx, decode any) *gomock
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPacketsByTraceTag", reflect.TypeOf((*MockQuerier)(nil).GetPacketsByTraceTag), ctx, decode)
|
||||
}
|
||||
|
||||
// GetPathStats mocks base method.
|
||||
func (m *MockQuerier) GetPathStats(ctx context.Context, arg db.GetPathStatsParams) ([]db.GetPathStatsRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "GetPathStats", ctx, arg)
|
||||
ret0, _ := ret[0].([]db.GetPathStatsRow)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// GetPathStats indicates an expected call of GetPathStats.
|
||||
func (mr *MockQuerierMockRecorder) GetPathStats(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPathStats", reflect.TypeOf((*MockQuerier)(nil).GetPathStats), ctx, arg)
|
||||
}
|
||||
|
||||
// GetRadioPresets mocks base method.
|
||||
func (m *MockQuerier) GetRadioPresets(ctx context.Context, arg db.GetRadioPresetsParams) ([]db.MvRadioPreset, error) {
|
||||
m.ctrl.T.Helper()
|
||||
@@ -1262,6 +1277,20 @@ func (mr *MockQuerierMockRecorder) RefreshObserverActivity(ctx any) *gomock.Call
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshObserverActivity", reflect.TypeOf((*MockQuerier)(nil).RefreshObserverActivity), ctx)
|
||||
}
|
||||
|
||||
// RefreshPathStats mocks base method.
|
||||
func (m *MockQuerier) RefreshPathStats(ctx context.Context) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "RefreshPathStats", ctx)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// RefreshPathStats indicates an expected call of RefreshPathStats.
|
||||
func (mr *MockQuerierMockRecorder) RefreshPathStats(ctx any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPathStats", reflect.TypeOf((*MockQuerier)(nil).RefreshPathStats), ctx)
|
||||
}
|
||||
|
||||
// RefreshPayloadBreakdown mocks base method.
|
||||
func (m *MockQuerier) RefreshPayloadBreakdown(ctx context.Context) error {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@@ -98,6 +98,15 @@ type MvObserverActivityHourly struct {
|
||||
RssiN int64 `json:"rssi_n"`
|
||||
}
|
||||
|
||||
type MvPathStatsHourly struct {
|
||||
Iata string `json:"iata"`
|
||||
Hour interface{} `json:"hour"`
|
||||
Category int32 `json:"category"`
|
||||
HashBytes int32 `json:"hash_bytes"`
|
||||
Entries int32 `json:"entries"`
|
||||
Receptions int64 `json:"receptions"`
|
||||
}
|
||||
|
||||
type MvPayloadBreakdownByIatum struct {
|
||||
Iata string `json:"iata"`
|
||||
PayloadType *int16 `json:"payload_type"`
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.31.1
|
||||
// source: paths.sql
|
||||
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
const getPathStats = `-- name: GetPathStats :many
|
||||
WITH readings AS (
|
||||
SELECT iata, hour, category, hash_bytes, entries, receptions FROM mv_path_stats_hourly
|
||||
WHERE hour >= $1::timestamptz AND hour < $2::timestamptz
|
||||
AND COALESCE(cardinality($3::bpchar[]), 0) = 0
|
||||
UNION ALL
|
||||
SELECT iata, hour, category, hash_bytes, entries, receptions FROM mv_path_stats_hourly
|
||||
WHERE hour >= $1::timestamptz AND hour < $2::timestamptz
|
||||
AND cardinality($3::bpchar[]) > 0 AND iata = ANY($3::bpchar[])
|
||||
)
|
||||
SELECT (grouping(hour) = 0)::boolean AS is_hourly,
|
||||
COALESCE(hour, 'epoch'::timestamptz)::timestamptz AS hour,
|
||||
category, hash_bytes, COALESCE(entries, 0)::integer AS entries,
|
||||
sum(receptions)::bigint AS receptions
|
||||
FROM readings
|
||||
GROUP BY GROUPING SETS ((category, hash_bytes, entries), (hour, category, hash_bytes))
|
||||
ORDER BY is_hourly, hour, category, hash_bytes, entries
|
||||
`
|
||||
|
||||
type GetPathStatsParams struct {
|
||||
Since pgtype.Timestamptz `json:"since"`
|
||||
Until pgtype.Timestamptz `json:"until"`
|
||||
Iatas []string `json:"iatas"`
|
||||
}
|
||||
|
||||
type GetPathStatsRow struct {
|
||||
IsHourly bool `json:"is_hourly"`
|
||||
Hour pgtype.Timestamptz `json:"hour"`
|
||||
Category int32 `json:"category"`
|
||||
HashBytes int32 `json:"hash_bytes"`
|
||||
Entries int32 `json:"entries"`
|
||||
Receptions int64 `json:"receptions"`
|
||||
}
|
||||
|
||||
// All HTTP aggregates read the compact classification snapshot.
|
||||
func (q *Queries) GetPathStats(ctx context.Context, arg GetPathStatsParams) ([]GetPathStatsRow, error) {
|
||||
rows, err := q.db.Query(ctx, getPathStats, arg.Since, arg.Until, arg.Iatas)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []GetPathStatsRow{}
|
||||
for rows.Next() {
|
||||
var i GetPathStatsRow
|
||||
if err := rows.Scan(
|
||||
&i.IsHourly,
|
||||
&i.Hour,
|
||||
&i.Category,
|
||||
&i.HashBytes,
|
||||
&i.Entries,
|
||||
&i.Receptions,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const refreshPathStats = `-- name: RefreshPathStats :exec
|
||||
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_path_stats_hourly
|
||||
`
|
||||
|
||||
func (q *Queries) RefreshPathStats(ctx context.Context) error {
|
||||
_, err := q.db.Exec(ctx, refreshPathStats)
|
||||
return err
|
||||
}
|
||||
@@ -79,6 +79,8 @@ type Querier interface {
|
||||
// Return distinct observation IATAs in first-heard order for path resolution,
|
||||
// without fetching full observations separately for every trace packet.
|
||||
GetPacketsByTraceTag(ctx context.Context, decode string) ([]GetPacketsByTraceTagRow, error)
|
||||
// All HTTP aggregates read the compact classification snapshot.
|
||||
GetPathStats(ctx context.Context, arg GetPathStatsParams) ([]GetPathStatsRow, error)
|
||||
GetRadioPresets(ctx context.Context, arg GetRadioPresetsParams) ([]MvRadioPreset, error)
|
||||
GetRegion(ctx context.Context, id int32) (GetRegionRow, error)
|
||||
GetRegionBySlug(ctx context.Context, slug string) (GetRegionBySlugRow, error)
|
||||
@@ -215,6 +217,7 @@ type Querier interface {
|
||||
ReconfirmRoutes(ctx context.Context, limit int32) error
|
||||
RefreshHourlyStats(ctx context.Context) error
|
||||
RefreshObserverActivity(ctx context.Context) error
|
||||
RefreshPathStats(ctx context.Context) error
|
||||
RefreshPayloadBreakdown(ctx context.Context) error
|
||||
RefreshRadioPresets(ctx context.Context) error
|
||||
RefreshSignalStats(ctx context.Context) error
|
||||
|
||||
Reference in new issue
Block a user