bgp: assemble BgpIntel from every source with per-source failures tolerated

This commit is contained in:
Andrew Pimenov 2026-08-09 02:16:59 +03:00
parent fee732dd8d
commit d45fcb5a59
2 changed files with 180 additions and 0 deletions

View file

@ -171,3 +171,135 @@ async def cached_json(
tmp.write_text(json.dumps(payload), encoding="utf-8") tmp.write_text(json.dumps(payload), encoding="utf-8")
os.replace(tmp, path) os.replace(tmp, path)
return payload return payload
from netcheck.models import BgpIntel
def build_bgp_intel(
asn: str,
overview: dict,
neighbours: dict,
prefixes: dict,
updates: dict,
asrank: dict | None,
pdb_net: dict | None,
pdb_ixlan: dict | None,
timeframe_days: int,
) -> BgpIntel:
info = parse_as_overview(overview)
upstreams, peers, downstreams = parse_asn_neighbours(neighbours)
announced, v4, v6 = parse_announced_prefixes(prefixes)
events = parse_bgp_updates(updates)
rank, cone_asns, cone_prefixes = parse_asrank(asrank) if asrank else (None, None, None)
info_type, traffic, _net_id = parse_peeringdb_net(pdb_net) if pdb_net else (None, None, None)
return BgpIntel(
asn=asn,
holder=info["holder"],
registry=info["registry"],
allocated_at=info["allocated_at"],
upstreams=upstreams,
peers=peers,
downstreams=downstreams,
announced_prefixes=announced,
prefix_count_v4=v4,
prefix_count_v6=v6,
flaps=events,
stability=classify_stability(events, timeframe_days),
ixps=parse_peeringdb_netixlan(pdb_ixlan) if pdb_ixlan else [],
pdb_info_type=info_type,
pdb_traffic=traffic,
asrank=rank,
cone_asns=cone_asns,
cone_prefixes=cone_prefixes,
)
import asyncio
_ASRANK_QUERY = """
query ASN($asn: String!) {
asn(asn: $asn) {
asn
rank
asnName
organization { orgName }
cone { numberAsns numberPrefixes numberAddresses }
asnDegree { provider peer customer total }
}
}
"""
async def collect_bgp(
client: httpx.AsyncClient,
providers: Providers,
cache_dir: Path,
asn: str,
peeringdb_key: str | None = None,
) -> tuple[BgpIntel, dict[str, object]]:
number = asn.upper().removeprefix("AS")
async def _asrank() -> dict:
response = await client.post(
providers.asrank_url, json={"query": _ASRANK_QUERY, "variables": {"asn": number}}
)
response.raise_for_status()
return response.json()
async def _pdb(endpoint: str, **params) -> dict:
headers = {"Authorization": f"Api-Key {peeringdb_key}"} if peeringdb_key else {}
response = await client.get(
f"{providers.peeringdb_base_url}/{endpoint}", params=params, headers=headers
)
response.raise_for_status()
return response.json()
settled = await asyncio.gather(
ripestat(client, providers, "as-overview", asn),
ripestat(client, providers, "asn-neighbours", asn),
ripestat(client, providers, "announced-prefixes", asn),
ripestat(client, providers, "bgp-updates", asn),
_asrank(),
cached_json(cache_dir, f"pdb-net-{number}", providers.peeringdb_cache_hours, lambda: _pdb("net", asn=number)),
return_exceptions=True,
)
overview, neighbours, prefixes, updates, asrank, pdb_net = [
None if isinstance(item, BaseException) else item for item in settled
]
pdb_ixlan = None
if pdb_net:
_, _, net_id = parse_peeringdb_net(pdb_net)
if net_id:
try:
pdb_ixlan = await cached_json(
cache_dir,
f"pdb-netixlan-{net_id}",
providers.peeringdb_cache_hours,
lambda: _pdb("netixlan", net_id=net_id),
)
except httpx.HTTPError:
pdb_ixlan = None
intel = build_bgp_intel(
asn=asn,
overview=overview or {},
neighbours=neighbours or {},
prefixes=prefixes or {},
updates=updates or {},
asrank=asrank,
pdb_net=pdb_net,
pdb_ixlan=pdb_ixlan,
timeframe_days=providers.ripestat_timeframe_days,
)
raw = {
"ripestat-as-overview": overview,
"ripestat-asn-neighbours": neighbours,
"ripestat-announced-prefixes": prefixes,
"ripestat-bgp-updates": updates,
"caida-asrank": asrank,
"peeringdb-net": pdb_net,
"peeringdb-netixlan": pdb_ixlan,
}
return intel, {k: v for k, v in raw.items() if v is not None}

View file

@ -192,3 +192,51 @@ async def test_cached_json_survives_a_corrupt_cache_file(tmp_path: Path):
return {"value": "fresh"} return {"value": "fresh"}
assert await cached_json(tmp_path, "k", ttl_hours=24, fetch=fetch) == {"value": "fresh"} assert await cached_json(tmp_path, "k", ttl_hours=24, fetch=fetch) == {"value": "fresh"}
from netcheck.bgp import build_bgp_intel
def test_build_bgp_intel_merges_every_source(api_fixture):
intel = build_bgp_intel(
asn="AS64500",
overview=api_fixture("ripestat_as_overview.json"),
neighbours=api_fixture("ripestat_asn_neighbours.json"),
prefixes=api_fixture("ripestat_announced_prefixes.json"),
updates=api_fixture("ripestat_bgp_updates.json"),
asrank=api_fixture("asrank.json"),
pdb_net=api_fixture("peeringdb_net.json"),
pdb_ixlan=api_fixture("peeringdb_netixlan.json"),
timeframe_days=14,
)
assert intel.asn == "AS64500"
assert intel.holder == "EXAMPLE-AS Example Telecom BV"
assert intel.upstreams == ["AS3356", "AS1299"]
assert intel.prefix_count_v4 == 2
assert intel.prefix_count_v6 == 1
assert intel.stability == "stable"
assert intel.asrank == 1842
assert intel.cone_asns == 37
assert intel.pdb_info_type == "Cable/DSL/ISP"
assert [i.name for i in intel.ixps] == ["AMS-IX", "DE-CIX Frankfurt"]
assert len(intel.flaps) == 3
def test_build_bgp_intel_tolerates_every_optional_source_being_absent(api_fixture):
intel = build_bgp_intel(
asn="AS64500",
overview=api_fixture("ripestat_as_overview.json"),
neighbours={},
prefixes={},
updates={},
asrank=None,
pdb_net=None,
pdb_ixlan=None,
timeframe_days=14,
)
assert intel.asn == "AS64500"
assert intel.upstreams == []
assert intel.asrank is None
assert intel.pdb_info_type is None
assert intel.ixps == []
assert intel.stability == "stable"