209 lines
8.5 KiB
Python
209 lines
8.5 KiB
Python
"""Collector baseline and coverage helpers for BGP observations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import case, distinct, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.time import to_iso8601_utc
|
|
from app.models.bgp_observation import BGPObservation
|
|
from app.services.collectors.bgp_common import RIPE_RIS_COLLECTOR_COORDS
|
|
|
|
|
|
def _collector_base_filters(source_filter: tuple[str, ...] | None) -> list[Any]:
|
|
filters: list[Any] = [
|
|
BGPObservation.collector.isnot(None),
|
|
func.length(func.btrim(BGPObservation.collector)) > 0,
|
|
]
|
|
if source_filter:
|
|
filters.append(BGPObservation.source.in_(source_filter))
|
|
return filters
|
|
|
|
|
|
async def build_bgp_collector_coverage(
|
|
db: AsyncSession,
|
|
*,
|
|
source_filter: tuple[str, ...] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
now = datetime.now(UTC)
|
|
recent_15m_threshold = now - timedelta(minutes=15)
|
|
recent_24h_threshold = now - timedelta(hours=24)
|
|
recent_7d_threshold = now - timedelta(days=7)
|
|
|
|
filters = _collector_base_filters(source_filter)
|
|
country_expr = func.nullif(BGPObservation.collector_geo["country"].astext, "")
|
|
city_expr = func.nullif(BGPObservation.collector_geo["city"].astext, "")
|
|
|
|
aggregate_stmt = (
|
|
select(
|
|
BGPObservation.collector.label("collector"),
|
|
func.count(BGPObservation.id).label("observation_count"),
|
|
func.count(distinct(BGPObservation.prefix)).label("prefix_count"),
|
|
func.count(distinct(BGPObservation.origin_asn)).label("origin_asn_count"),
|
|
func.count(distinct(BGPObservation.peer_asn)).label("peer_asn_count"),
|
|
func.sum(case((BGPObservation.observed_at >= recent_15m_threshold, 1), else_=0)).label("recent_15m_observation_count"),
|
|
func.sum(case((BGPObservation.observed_at >= recent_24h_threshold, 1), else_=0)).label("recent_24h_observation_count"),
|
|
func.sum(case((BGPObservation.observed_at >= recent_7d_threshold, 1), else_=0)).label("recent_7d_observation_count"),
|
|
func.count(distinct(case((BGPObservation.observed_at >= recent_15m_threshold, BGPObservation.prefix), else_=None))).label("recent_15m_prefix_count"),
|
|
func.count(distinct(case((BGPObservation.observed_at >= recent_24h_threshold, BGPObservation.prefix), else_=None))).label("recent_24h_prefix_count"),
|
|
func.count(distinct(case((BGPObservation.observed_at >= recent_7d_threshold, BGPObservation.prefix), else_=None))).label("recent_7d_prefix_count"),
|
|
func.max(BGPObservation.observed_at).label("latest_observed_at"),
|
|
)
|
|
.where(*filters)
|
|
.group_by(BGPObservation.collector)
|
|
)
|
|
aggregate_rows = (await db.execute(aggregate_stmt)).all()
|
|
|
|
latest_subquery = (
|
|
select(
|
|
BGPObservation.collector.label("collector"),
|
|
BGPObservation.event_type.label("latest_event_type"),
|
|
country_expr.label("country"),
|
|
city_expr.label("city"),
|
|
func.row_number()
|
|
.over(
|
|
partition_by=BGPObservation.collector,
|
|
order_by=(BGPObservation.observed_at.desc(), BGPObservation.id.desc()),
|
|
)
|
|
.label("rn"),
|
|
)
|
|
.where(*filters)
|
|
.subquery()
|
|
)
|
|
latest_rows = (
|
|
await db.execute(
|
|
select(
|
|
latest_subquery.c.collector,
|
|
latest_subquery.c.latest_event_type,
|
|
latest_subquery.c.country,
|
|
latest_subquery.c.city,
|
|
).where(latest_subquery.c.rn == 1)
|
|
)
|
|
).all()
|
|
|
|
event_counts_subquery = (
|
|
select(
|
|
BGPObservation.collector.label("collector"),
|
|
BGPObservation.event_type.label("event_type"),
|
|
func.count(BGPObservation.id).label("count"),
|
|
func.row_number()
|
|
.over(
|
|
partition_by=BGPObservation.collector,
|
|
order_by=(func.count(BGPObservation.id).desc(), BGPObservation.event_type.asc()),
|
|
)
|
|
.label("rn"),
|
|
)
|
|
.where(*filters)
|
|
.group_by(BGPObservation.collector, BGPObservation.event_type)
|
|
.subquery()
|
|
)
|
|
top_event_rows = (
|
|
await db.execute(
|
|
select(
|
|
event_counts_subquery.c.collector,
|
|
event_counts_subquery.c.event_type,
|
|
event_counts_subquery.c.count,
|
|
).where(event_counts_subquery.c.rn <= 3)
|
|
)
|
|
).all()
|
|
|
|
scope_rows = (
|
|
await db.execute(
|
|
select(
|
|
BGPObservation.collector.label("collector"),
|
|
country_expr.label("country"),
|
|
city_expr.label("city"),
|
|
)
|
|
.where(*filters)
|
|
.distinct()
|
|
)
|
|
).all()
|
|
|
|
latest_by_collector = {
|
|
row.collector: {
|
|
"latest_event_type": row.latest_event_type,
|
|
"country": row.country,
|
|
"city": row.city,
|
|
}
|
|
for row in latest_rows
|
|
}
|
|
|
|
scope_by_collector: dict[str, dict[str, set[str]]] = defaultdict(lambda: {"countries": set(), "cities": set()})
|
|
for row in scope_rows:
|
|
if row.country:
|
|
scope_by_collector[row.collector]["countries"].add(row.country)
|
|
if row.city:
|
|
scope_by_collector[row.collector]["cities"].add(row.city)
|
|
|
|
top_events_by_collector: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for row in top_event_rows:
|
|
top_events_by_collector[row.collector].append(
|
|
{"event_type": row.event_type, "count": row.count}
|
|
)
|
|
|
|
by_collector: dict[str, dict[str, Any]] = {}
|
|
for row in aggregate_rows:
|
|
collector = row.collector
|
|
latest = latest_by_collector.get(collector, {})
|
|
fallback_location = RIPE_RIS_COLLECTOR_COORDS.get(collector, {})
|
|
scope = scope_by_collector.get(collector, {"countries": set(), "cities": set()})
|
|
|
|
by_collector[collector] = {
|
|
"collector": collector,
|
|
"city": latest.get("city") or fallback_location.get("city"),
|
|
"country": latest.get("country") or fallback_location.get("country"),
|
|
"latitude": fallback_location.get("latitude"),
|
|
"longitude": fallback_location.get("longitude"),
|
|
"observation_count": row.observation_count or 0,
|
|
"prefix_count": row.prefix_count or 0,
|
|
"origin_asn_count": row.origin_asn_count or 0,
|
|
"peer_asn_count": row.peer_asn_count or 0,
|
|
"recent_15m_observation_count": row.recent_15m_observation_count or 0,
|
|
"recent_24h_observation_count": row.recent_24h_observation_count or 0,
|
|
"recent_7d_observation_count": row.recent_7d_observation_count or 0,
|
|
"recent_15m_prefix_count": row.recent_15m_prefix_count or 0,
|
|
"recent_24h_prefix_count": row.recent_24h_prefix_count or 0,
|
|
"recent_7d_prefix_count": row.recent_7d_prefix_count or 0,
|
|
"top_event_types": top_events_by_collector.get(collector, []),
|
|
"latest_observed_at": to_iso8601_utc(row.latest_observed_at),
|
|
"latest_event_type": latest.get("latest_event_type"),
|
|
"baseline_scope": {
|
|
"countries": sorted(scope["countries"]),
|
|
"cities": sorted(scope["cities"]),
|
|
},
|
|
}
|
|
|
|
for collector, location in RIPE_RIS_COLLECTOR_COORDS.items():
|
|
if collector in by_collector:
|
|
continue
|
|
by_collector[collector] = {
|
|
"collector": collector,
|
|
"city": location.get("city"),
|
|
"country": location.get("country"),
|
|
"latitude": location.get("latitude"),
|
|
"longitude": location.get("longitude"),
|
|
"observation_count": 0,
|
|
"prefix_count": 0,
|
|
"origin_asn_count": 0,
|
|
"peer_asn_count": 0,
|
|
"recent_15m_observation_count": 0,
|
|
"recent_24h_observation_count": 0,
|
|
"recent_7d_observation_count": 0,
|
|
"recent_15m_prefix_count": 0,
|
|
"recent_24h_prefix_count": 0,
|
|
"recent_7d_prefix_count": 0,
|
|
"top_event_types": [],
|
|
"latest_observed_at": None,
|
|
"latest_event_type": None,
|
|
"baseline_scope": {
|
|
"countries": [location["country"]] if location.get("country") else [],
|
|
"cities": [location["city"]] if location.get("city") else [],
|
|
},
|
|
}
|
|
|
|
return [by_collector[collector] for collector in sorted(by_collector.keys())]
|