mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-22 22:21:27 +00:00
feat(stats): filter scope counts by IATA and region (#142)
* feat(stats): filter scope counts by IATA and region * perf(stats): aggregate region observations before counting scopes
This commit is contained in:
1 parent
b21279c747
commit
bcdea76c26
18 files changed
+476
-34
No files matched your search
@@ -653,18 +653,18 @@ func (mr *MockQuerierMockRecorder) GetScopeNames(ctx any) *gomock.Call {
|
||||
}
|
||||
|
||||
// GetScopeStats mocks base method.
|
||||
func (m *MockQuerier) GetScopeStats(ctx context.Context) ([]db.GetScopeStatsRow, error) {
|
||||
func (m *MockQuerier) GetScopeStats(ctx context.Context, iatas []string) ([]db.GetScopeStatsRow, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "GetScopeStats", ctx)
|
||||
ret := m.ctrl.Call(m, "GetScopeStats", ctx, iatas)
|
||||
ret0, _ := ret[0].([]db.GetScopeStatsRow)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// GetScopeStats indicates an expected call of GetScopeStats.
|
||||
func (mr *MockQuerierMockRecorder) GetScopeStats(ctx any) *gomock.Call {
|
||||
func (mr *MockQuerierMockRecorder) GetScopeStats(ctx, iatas any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeStats", reflect.TypeOf((*MockQuerier)(nil).GetScopeStats), ctx)
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeStats", reflect.TypeOf((*MockQuerier)(nil).GetScopeStats), ctx, iatas)
|
||||
}
|
||||
|
||||
// GetScopesByIATAs mocks base method.
|
||||
|
||||
+4
-3
@@ -82,9 +82,10 @@ type Querier interface {
|
||||
GetRegionIATAs(ctx context.Context, regionID int32) ([]string, error)
|
||||
GetScopeByName(ctx context.Context, name string) (GetScopeByNameRow, error)
|
||||
GetScopeNames(ctx context.Context) ([]string, error)
|
||||
// Count each table on its own; the old cross-join blew up to millions of rows
|
||||
// before COUNT(DISTINCT) (~10s).
|
||||
GetScopeStats(ctx context.Context) ([]GetScopeStatsRow, error)
|
||||
// Aggregate matching observations once, separately from node memberships to avoid
|
||||
// a cross-join. Empty IATAs keep the original global counts, including associations
|
||||
// whose observations have expired; the filtered aggregates are empty in that case.
|
||||
GetScopeStats(ctx context.Context, iatas []string) ([]GetScopeStatsRow, error)
|
||||
GetScopesByIATAs(ctx context.Context, dollar_1 []string) ([]GetScopesByIATAsRow, error)
|
||||
// Repeaters/room servers (node_type 2/3) whose current advert-derived clock drift exceeds
|
||||
// the given threshold in magnitude, worst first. Not time-windowed -- reflects each node's
|
||||
|
||||
+31
-7
@@ -1418,12 +1418,35 @@ func (q *Queries) GetScopeNames(ctx context.Context) ([]string, error) {
|
||||
}
|
||||
|
||||
const getScopeStats = `-- name: GetScopeStats :many
|
||||
WITH observation_counts AS (
|
||||
SELECT p.scope_id,
|
||||
COUNT(DISTINCT p.packet_hash) AS packet_count,
|
||||
COUNT(DISTINCT os.observer_id) AS observer_count
|
||||
FROM packet_observations po
|
||||
JOIN packets p ON p.packet_hash = po.packet_hash
|
||||
LEFT JOIN observer_scopes os ON os.scope_id = p.scope_id AND os.observer_id = po.observer_id
|
||||
WHERE po.iata = ANY($1::bpchar[]) AND p.scope_id IS NOT NULL
|
||||
GROUP BY p.scope_id
|
||||
), node_counts AS (
|
||||
SELECT n.default_scope_id AS scope_id, COUNT(*) AS node_count
|
||||
FROM nodes n
|
||||
WHERE n.id IN (SELECT node_id FROM node_iatas WHERE iata = ANY($1::bpchar[]))
|
||||
GROUP BY n.default_scope_id
|
||||
)
|
||||
SELECT
|
||||
ts.name,
|
||||
(SELECT COUNT(*) FROM packets p WHERE p.scope_id = ts.id) AS packet_count,
|
||||
(SELECT COUNT(*) FROM observer_scopes os WHERE os.scope_id = ts.id) AS observer_count,
|
||||
(SELECT COUNT(*) FROM nodes n WHERE n.default_scope_id = ts.id) AS node_count
|
||||
CASE WHEN COALESCE(cardinality($1::bpchar[]), 0) = 0
|
||||
THEN (SELECT COUNT(*) FROM packets p WHERE p.scope_id = ts.id)
|
||||
ELSE COALESCE(oc.packet_count, 0) END::bigint AS packet_count,
|
||||
CASE WHEN COALESCE(cardinality($1::bpchar[]), 0) = 0
|
||||
THEN (SELECT COUNT(*) FROM observer_scopes os WHERE os.scope_id = ts.id)
|
||||
ELSE COALESCE(oc.observer_count, 0) END::bigint AS observer_count,
|
||||
CASE WHEN COALESCE(cardinality($1::bpchar[]), 0) = 0
|
||||
THEN (SELECT COUNT(*) FROM nodes n WHERE n.default_scope_id = ts.id)
|
||||
ELSE COALESCE(nc.node_count, 0) END::bigint AS node_count
|
||||
FROM transport_scopes ts
|
||||
LEFT JOIN observation_counts oc ON oc.scope_id = ts.id
|
||||
LEFT JOIN node_counts nc ON nc.scope_id = ts.id
|
||||
ORDER BY ts.name
|
||||
`
|
||||
|
||||
@@ -1434,10 +1457,11 @@ type GetScopeStatsRow struct {
|
||||
NodeCount int64 `json:"node_count"`
|
||||
}
|
||||
|
||||
// Count each table on its own; the old cross-join blew up to millions of rows
|
||||
// before COUNT(DISTINCT) (~10s).
|
||||
func (q *Queries) GetScopeStats(ctx context.Context) ([]GetScopeStatsRow, error) {
|
||||
rows, err := q.db.Query(ctx, getScopeStats)
|
||||
// Aggregate matching observations once, separately from node memberships to avoid
|
||||
// a cross-join. Empty IATAs keep the original global counts, including associations
|
||||
// whose observations have expired; the filtered aggregates are empty in that case.
|
||||
func (q *Queries) GetScopeStats(ctx context.Context, iatas []string) ([]GetScopeStatsRow, error) {
|
||||
rows, err := q.db.Query(ctx, getScopeStats, iatas)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user