linkong/planet
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: expand bgp observability surfaces

Browse Source
This commit is contained in:
linkong
2026-03-31 14:07:28 +08:00
parent ac63bba2a2
commit 552e49bde0
16 changed files with 1319 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend/app/api/v1/bgp.py
View File

@@ -11,6 +11,7 @@ from app.models.bgp_anomaly import BGPAnomaly
from app.models.bgp_incident import BGPIncident
from app.models.bgp_observation import BGPObservation
from app.models.user import User
from app.services.bgp_collectors import build_bgp_collector_coverage
router = APIRouter()
@@ -108,6 +109,35 @@ async def get_bgp_event_summary(
}
@router.get("/collectors")
async def list_bgp_collectors(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
data = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
return {
"total": len(data),
"data": data,
}
@router.get("/collectors/summary")
async def get_bgp_collector_summary(
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
collectors = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
active_collectors = [item for item in collectors if item["observation_count"] > 0]
return {
"total": len(collectors),
"active_collectors": len(active_collectors),
"observed_prefixes": sum(item["prefix_count"] for item in active_collectors),
"observed_origins": sum(item["origin_asn_count"] for item in active_collectors),
"recent_24h_events": sum(item["recent_24h_observation_count"] for item in active_collectors),
"recent_7d_events": sum(item["recent_7d_observation_count"] for item in active_collectors),
}
@router.get("/events/{event_id}")
async def get_bgp_event(
event_id: int,

42
backend/app/api/v1/visualization.py
View File

@@ -17,6 +17,7 @@ from app.db.session import get_db
from app.models.bgp_anomaly import BGPAnomaly
from app.models.bgp_incident import BGPIncident
from app.models.collected_data import CollectedData
from app.services.bgp_collectors import build_bgp_collector_coverage
from app.services.cable_graph import build_graph_from_data, CableGraph
from app.services.collectors.bgp_common import RIPE_RIS_COLLECTOR_COORDS
@@ -417,10 +418,14 @@ def convert_bgp_anomalies_to_geojson(records: List[BGPAnomaly]) -> Dict[str, Any
return {"type": "FeatureCollection", "features": features}
def convert_bgp_collectors_to_geojson() -> Dict[str, Any]:
def convert_bgp_collectors_to_geojson(
coverage_by_collector: Dict[str, Dict[str, Any]] | None = None,
) -> Dict[str, Any]:
features = []
coverage_by_collector = coverage_by_collector or {}
for collector, location in sorted(RIPE_RIS_COLLECTOR_COORDS.items()):
coverage = coverage_by_collector.get(collector, {})
features.append(
{
"type": "Feature",
@@ -430,9 +435,27 @@ def convert_bgp_collectors_to_geojson() -> Dict[str, Any]:
},
"properties": {
"collector": collector,
"city": location.get("city"),
"country": location.get("country"),
"city": coverage.get("city") or location.get("city"),
"country": coverage.get("country") or location.get("country"),
"status": "online",
"observation_count": coverage.get("observation_count", 0),
"prefix_count": coverage.get("prefix_count", 0),
"origin_asn_count": coverage.get("origin_asn_count", 0),
"peer_asn_count": coverage.get("peer_asn_count", 0),
"recent_24h_observation_count": coverage.get("recent_24h_observation_count", 0),
"recent_7d_observation_count": coverage.get("recent_7d_observation_count", 0),
"recent_24h_prefix_count": coverage.get("recent_24h_prefix_count", 0),
"recent_7d_prefix_count": coverage.get("recent_7d_prefix_count", 0),
"top_event_types": coverage.get("top_event_types", []),
"latest_observed_at": coverage.get("latest_observed_at"),
"latest_event_type": coverage.get("latest_event_type"),
"baseline_scope": coverage.get(
"baseline_scope",
{
"countries": [location.get("country")] if location.get("country") else [],
"cities": [location.get("city")] if location.get("city") else [],
},
),
},
}
)
@@ -740,8 +763,17 @@ async def get_bgp_incidents_geojson(
@router.get("/geo/bgp-collectors")
async def get_bgp_collectors_geojson():
geojson = convert_bgp_collectors_to_geojson()
async def get_bgp_collectors_geojson(db: AsyncSession = Depends(get_db)):
coverage = await build_bgp_collector_coverage(
db,
source_filter=("ris_live_bgp", "bgpstream_bgp"),
)
coverage_by_collector = {
item["collector"]: item
for item in coverage
if item.get("collector")
}
geojson = convert_bgp_collectors_to_geojson(coverage_by_collector)
return {**geojson, "count": len(geojson.get("features", []))}

162
backend/app/services/bgp_collectors.py Normal file
View File

@@ -0,0 +1,162 @@
"""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 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
async def build_bgp_collector_coverage(
db: AsyncSession,
*,
source_filter: tuple[str, ...] | None = None,
) -> list[dict[str, Any]]:
now = datetime.now(UTC)
recent_24h_threshold = now - timedelta(hours=24)
recent_7d_threshold = now - timedelta(days=7)
stmt = select(BGPObservation).order_by(BGPObservation.observed_at.desc(), BGPObservation.id.desc())
if source_filter:
stmt = stmt.where(BGPObservation.source.in_(source_filter))
result = await db.execute(stmt)
records = list(result.scalars().all())
by_collector: dict[str, dict[str, Any]] = {}
for record in records:
collector = str(record.collector or "").strip()
if not collector:
continue
coverage = by_collector.get(collector)
if coverage is None:
location = record.collector_geo or RIPE_RIS_COLLECTOR_COORDS.get(collector, {})
coverage = {
"collector": collector,
"city": location.get("city"),
"country": location.get("country"),
"latitude": location.get("latitude"),
"longitude": location.get("longitude"),
"observation_count": 0,
"prefixes": set(),
"origin_asns": set(),
"peer_asns": set(),
"event_types": defaultdict(int),
"countries": set(),
"cities": set(),
"recent_24h_observation_count": 0,
"recent_7d_observation_count": 0,
"recent_24h_prefixes": set(),
"recent_7d_prefixes": set(),
"latest_observed_at": None,
"latest_event_type": None,
}
by_collector[collector] = coverage
coverage["observation_count"] += 1
if record.prefix:
coverage["prefixes"].add(record.prefix)
if record.origin_asn is not None:
coverage["origin_asns"].add(record.origin_asn)
if record.peer_asn is not None:
coverage["peer_asns"].add(record.peer_asn)
if record.event_type:
coverage["event_types"][record.event_type] += 1
observed_at = record.observed_at
if observed_at is not None:
aware_observed_at = (
observed_at.astimezone(UTC)
if observed_at.tzinfo
else observed_at.replace(tzinfo=UTC)
)
if aware_observed_at >= recent_24h_threshold:
coverage["recent_24h_observation_count"] += 1
if record.prefix:
coverage["recent_24h_prefixes"].add(record.prefix)
if aware_observed_at >= recent_7d_threshold:
coverage["recent_7d_observation_count"] += 1
if record.prefix:
coverage["recent_7d_prefixes"].add(record.prefix)
geo = record.collector_geo or {}
if geo.get("country"):
coverage["countries"].add(geo["country"])
if geo.get("city"):
coverage["cities"].add(geo["city"])
current_latest = coverage["latest_observed_at"]
if current_latest is None or (
record.observed_at is not None and record.observed_at > current_latest
):
coverage["latest_observed_at"] = record.observed_at
coverage["latest_event_type"] = record.event_type
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,
"prefixes": set(),
"origin_asns": set(),
"peer_asns": set(),
"event_types": defaultdict(int),
"countries": {location.get("country")} if location.get("country") else set(),
"cities": {location.get("city")} if location.get("city") else set(),
"recent_24h_observation_count": 0,
"recent_7d_observation_count": 0,
"recent_24h_prefixes": set(),
"recent_7d_prefixes": set(),
"latest_observed_at": None,
"latest_event_type": None,
}
results: list[dict[str, Any]] = []
for collector in sorted(by_collector.keys()):
item = by_collector[collector]
top_event_types = sorted(
item["event_types"].items(),
key=lambda pair: (-pair[1], pair[0]),
)
results.append(
{
"collector": item["collector"],
"city": item["city"],
"country": item["country"],
"latitude": item["latitude"],
"longitude": item["longitude"],
"observation_count": item["observation_count"],
"prefix_count": len(item["prefixes"]),
"origin_asn_count": len(item["origin_asns"]),
"peer_asn_count": len(item["peer_asns"]),
"recent_24h_observation_count": item["recent_24h_observation_count"],
"recent_7d_observation_count": item["recent_7d_observation_count"],
"recent_24h_prefix_count": len(item["recent_24h_prefixes"]),
"recent_7d_prefix_count": len(item["recent_7d_prefixes"]),
"top_event_types": [
{"event_type": event_type, "count": count}
for event_type, count in top_event_types[:3]
],
"latest_observed_at": to_iso8601_utc(item["latest_observed_at"]),
"latest_event_type": item["latest_event_type"],
"baseline_scope": {
"countries": sorted(country for country in item["countries"] if country),
"cities": sorted(city for city in item["cities"] if city),
},
}
)
return results

153
backend/app/services/bgp_incidents.py
View File

@@ -3,12 +3,16 @@
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.collected_data_fields import get_record_field
from app.models.bgp_anomaly import BGPAnomaly
from app.models.bgp_incident import BGPIncident
from app.models.collected_data import CollectedData
from app.services.cable_graph import haversine_distance
def _severity_rank(value: str | None) -> int:
@@ -44,6 +48,150 @@ def _collector_regions_from_anomaly(anomaly: BGPAnomaly) -> list[dict]:
return collected
def _dedupe_collected_records(records: list[CollectedData]) -> list[CollectedData]:
latest_by_key: dict[str, CollectedData] = {}
for record in records:
dedupe_key = str(record.source_id or record.entity_key or record.name or record.id)
existing = latest_by_key.get(dedupe_key)
if existing is None or (record.id or 0) > (existing.id or 0):
latest_by_key[dedupe_key] = record
return list(latest_by_key.values())
async def infer_related_infrastructure(
db: AsyncSession,
affected_regions: list[dict],
*,
max_matches: int = 6,
max_distance_km: float = 450.0,
) -> dict[str, list[dict[str, Any]]]:
valid_regions = [
region
for region in affected_regions
if isinstance(region, dict)
and isinstance(region.get("latitude"), (int, float))
and isinstance(region.get("longitude"), (int, float))
]
if not valid_regions:
return {"related_cables": [], "related_ixps": []}
landing_result = await db.execute(
select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
)
relation_result = await db.execute(
select(CollectedData).where(CollectedData.source == "arcgis_cable_landing_relation")
)
cable_result = await db.execute(
select(CollectedData).where(CollectedData.source == "arcgis_cables")
)
landing_records = _dedupe_collected_records(list(landing_result.scalars().all()))
relation_records = _dedupe_collected_records(list(relation_result.scalars().all()))
cable_records = _dedupe_collected_records(list(cable_result.scalars().all()))
city_to_cable_ids: dict[int, list[int]] = {}
for relation in relation_records:
metadata = relation.extra_data or {}
city_id = metadata.get("city_id")
cable_id = metadata.get("cable_id")
if city_id is None or cable_id is None:
continue
city_key = int(city_id)
cable_key = int(cable_id)
city_to_cable_ids.setdefault(city_key, [])
if cable_key not in city_to_cable_ids[city_key]:
city_to_cable_ids[city_key].append(cable_key)
cable_id_to_name: dict[int, str] = {}
for cable in cable_records:
metadata = cable.extra_data or {}
cable_id = metadata.get("cable_id")
if cable_id is None or not cable.name:
continue
cable_id_to_name[int(cable_id)] = cable.name
matches: list[dict[str, Any]] = []
seen_match_keys: set[tuple[Any, ...]] = set()
for region in valid_regions:
region_coords = (float(region["longitude"]), float(region["latitude"]))
for landing in landing_records:
try:
latitude = get_record_field(landing, "latitude")
longitude = get_record_field(landing, "longitude")
landing_lat = float(latitude) if latitude is not None else None
landing_lon = float(longitude) if longitude is not None else None
except (TypeError, ValueError):
landing_lat = None
landing_lon = None
if landing_lat is None or landing_lon is None:
continue
distance_km = haversine_distance(region_coords, (landing_lon, landing_lat))
if distance_km > max_distance_km:
continue
landing_meta = landing.extra_data or {}
city_id = landing_meta.get("city_id")
cable_names = []
if city_id is not None:
for cable_id in city_to_cable_ids.get(int(city_id), []):
cable_name = cable_id_to_name.get(int(cable_id))
if cable_name and cable_name not in cable_names:
cable_names.append(cable_name)
match = {
"landing_point": landing.name or "Unknown",
"city": get_record_field(landing, "city"),
"country": get_record_field(landing, "country"),
"distance_km": round(distance_km, 1),
"collector": region.get("collector"),
"cable_names": cable_names,
}
match_key = (
match["landing_point"],
match["city"],
match["country"],
)
if match_key in seen_match_keys:
continue
seen_match_keys.add(match_key)
matches.append(match)
matches.sort(
key=lambda item: (
item.get("distance_km", 999999),
str(item.get("landing_point") or ""),
)
)
matches = matches[:max_matches]
related_ixps = []
seen_ixp_keys: set[tuple[str, str]] = set()
for item in matches:
city = str(item.get("city") or "").strip()
country = str(item.get("country") or "").strip()
if not city and not country:
continue
key = (city, country)
if key in seen_ixp_keys:
continue
seen_ixp_keys.add(key)
related_ixps.append(
{
"name": ", ".join(part for part in [city, country] if part),
"type": "regional_exchange_hint",
}
)
return {
"related_cables": matches,
"related_ixps": related_ixps,
}
async def create_bgp_incidents_for_anomalies(
db: AsyncSession,
*,
@@ -120,6 +268,7 @@ async def create_bgp_incidents_for_anomalies(
f"{len(items)} anomaly signal(s) grouped into one {primary.anomaly_type} incident, "
f"affecting {len(prefixes) or 1} prefix scope(s) across {len(collectors)} collector(s)."
)
related_infrastructure = await infer_related_infrastructure(db, regions)
db.add(
BGPIncident(
@@ -138,8 +287,8 @@ async def create_bgp_incidents_for_anomalies(
affected_asns=asns,
affected_collectors=collectors,
affected_regions=regions,
related_cables=[],
related_ixps=[],
related_cables=related_infrastructure["related_cables"],
related_ixps=related_infrastructure["related_ixps"],
evidence_refs=evidence_refs,
)
)

147
backend/tests/test_bgp.py
View File

@@ -16,7 +16,11 @@ from app.services.collectors.bgp_common import (
save_bgp_observations_for_batch,
)
from app.services.bgp_enrichment import enrich_bgp_events_for_batch, extract_bgp_network_fields
from app.services.bgp_incidents import create_bgp_incidents_for_anomalies
from app.services.bgp_incidents import (
create_bgp_incidents_for_anomalies,
infer_related_infrastructure,
)
from app.services.bgp_collectors import build_bgp_collector_coverage
from app.models.bgp_anomaly import BGPAnomaly
from app.models.collected_data import CollectedData
from app.models.bgp_incident import BGPIncident
@@ -331,13 +335,17 @@ async def test_create_bgp_incidents_for_anomalies_aggregates_regions_and_collect
},
)
created = await create_bgp_incidents_for_anomalies(
db,
source="ris_live_bgp",
snapshot_id=1,
task_id=2,
anomalies=[anomaly],
)
with patch(
"app.services.bgp_incidents.infer_related_infrastructure",
new=AsyncMock(return_value={"related_cables": [], "related_ixps": []}),
):
created = await create_bgp_incidents_for_anomalies(
db,
source="ris_live_bgp",
snapshot_id=1,
task_id=2,
anomalies=[anomaly],
)
assert created == 1
assert db.commits == 1
@@ -348,6 +356,91 @@ async def test_create_bgp_incidents_for_anomalies_aggregates_regions_and_collect
assert incident.affected_regions[0]["city"] == "Amsterdam"
@pytest.mark.asyncio
async def test_infer_related_infrastructure_links_nearby_cables():
landing = CollectedData(
source="arcgis_landing_points",
name="Amsterdam Landing",
data_type="landing_point",
extra_data={
"city_id": 10,
"country": "Netherlands",
"city": "Amsterdam",
"latitude": 52.3676,
"longitude": 4.9041,
},
)
relation = CollectedData(
source="arcgis_cable_landing_relation",
name="rel-1",
data_type="landing_relation",
extra_data={"city_id": 10, "cable_id": 20},
)
cable = CollectedData(
source="arcgis_cables",
name="AEConnect-1",
data_type="cable",
extra_data={"cable_id": 20},
)
db = _FakeAsyncSession([[landing], [relation], [cable]])
result = await infer_related_infrastructure(
db,
[
{
"collector": "rrc00",
"country": "Netherlands",
"city": "Amsterdam",
"latitude": 52.36,
"longitude": 4.90,
}
],
)
assert len(result["related_cables"]) == 1
assert result["related_cables"][0]["landing_point"] == "Amsterdam Landing"
assert result["related_cables"][0]["cable_names"] == ["AEConnect-1"]
assert result["related_ixps"][0]["name"] == "Amsterdam, Netherlands"
@pytest.mark.asyncio
async def test_build_bgp_collector_coverage_summarizes_observations():
obs_one = BGPObservation(
source="ris_live_bgp",
collector="rrc00",
prefix="203.0.113.0/24",
origin_asn=64496,
peer_asn=3333,
event_type="announcement",
observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC),
collector_geo={"city": "Amsterdam", "country": "Netherlands"},
)
obs_two = BGPObservation(
source="ris_live_bgp",
collector="rrc00",
prefix="198.51.100.0/24",
origin_asn=64497,
peer_asn=3334,
event_type="withdrawal",
observed_at=datetime(2026, 3, 30, 10, 5, tzinfo=UTC),
collector_geo={"city": "Amsterdam", "country": "Netherlands"},
)
db = _FakeAsyncSession([[obs_one, obs_two]])
coverage = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
first = next(item for item in coverage if item["collector"] == "rrc00")
assert first["observation_count"] == 2
assert first["recent_24h_observation_count"] == 2
assert first["recent_7d_observation_count"] == 2
assert first["prefix_count"] == 2
assert first["recent_24h_prefix_count"] == 2
assert first["origin_asn_count"] == 2
assert first["latest_event_type"] == "withdrawal"
assert first["baseline_scope"]["countries"] == ["Netherlands"]
assert first["baseline_scope"]["cities"] == ["Amsterdam"]
@pytest.mark.asyncio
async def test_save_bgp_observations_for_batch_adds_rows():
db = _FakeAsyncSession([])
@@ -643,3 +736,41 @@ async def test_bgp_event_summary_api_returns_aggregates():
assert payload["prefix_count"] == 2
assert payload["by_type"]["announcement"] == 1
assert payload["by_type"]["withdrawal"] == 1
@pytest.mark.asyncio
async def test_bgp_collectors_api_returns_coverage():
observation = BGPObservation(
id=1,
source="ris_live_bgp",
collector="rrc00",
peer_asn=3333,
prefix="203.0.113.0/24",
event_type="announcement",
origin_asn=64496,
observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC),
collector_geo={"city": "Amsterdam", "country": "Netherlands"},
)
db = _FakeAsyncSession([[observation], [observation]])
client = await _bgp_test_client(db)
try:
list_response = await client.get("/api/v1/bgp/collectors")
summary_response = await client.get("/api/v1/bgp/collectors/summary")
finally:
await client.aclose()
app.dependency_overrides.clear()
assert list_response.status_code == 200
list_payload = list_response.json()
assert list_payload["total"] >= 1
target = next(item for item in list_payload["data"] if item["collector"] == "rrc00")
assert target["observation_count"] == 1
assert target["prefix_count"] == 1
assert summary_response.status_code == 200
summary_payload = summary_response.json()
assert summary_payload["active_collectors"] >= 1
assert summary_payload["observed_prefixes"] >= 1
assert summary_payload["recent_24h_events"] >= 1
assert summary_payload["recent_7d_events"] >= 1