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

feat(bgp): improve prefix-geo pipeline and collector reliability

Browse Source
This commit is contained in:
linkong
2026-04-07 15:23:26 +08:00
parent 5b9ef0223d
commit c439e91d12
11 changed files with 621 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
backend/app/api/v1/datasources.py
View File

@@ -17,6 +17,7 @@ from app.models.user import User
from app.services.scheduler import get_latest_task_id_for_datasource, run_collector_now, sync_datasource_job
router = APIRouter()
STALE_RUNNING_TASK_TIMEOUT_MINUTES = 90
def format_frequency_label(minutes: int) -> str:
@@ -71,7 +72,32 @@ async def get_running_task(db: AsyncSession, datasource_id: int) -> Optional[Col
.order_by(CollectionTask.started_at.desc())
.limit(1)
)
return result.scalar_one_or_none()
task = result.scalar_one_or_none()
if not task:
return None
started_at = task.started_at
if started_at is None:
return task
now = datetime.now(timezone.utc)
if started_at.tzinfo is None:
started_at = started_at.replace(tzinfo=timezone.utc)
if now - started_at <= timedelta(minutes=STALE_RUNNING_TASK_TIMEOUT_MINUTES):
return task
existing_error = (task.error_message or "").strip()
stale_reason = (
f"Marked failed automatically after stale running timeout "
f"({STALE_RUNNING_TASK_TIMEOUT_MINUTES}m)"
)
task.status = "failed"
task.phase = "failed"
task.completed_at = now
task.error_message = f"{existing_error}\n{stale_reason}".strip() if existing_error else stale_reason
await db.commit()
return None
@router.get("")

2
backend/app/core/data_sources.py
View File

@@ -26,6 +26,8 @@ COLLECTOR_URL_KEYS = {
"ris_live_bgp": "ris_live.url",
"bgpstream_bgp": "bgpstream.url",
"iptoasn_prefix_geo": "iptoasn.combined_url",
"opengeofeed_prefix_geo": "opengeofeed.public_csv_url",
"nro_delegated_prefix_geo": "nro.delegated_stats_url",
}

6
backend/app/core/data_sources.yaml
View File

@@ -46,3 +46,9 @@ bgpstream:
iptoasn:
combined_url: "https://iptoasn.com/data/ip2asn-combined.tsv.gz"
opengeofeed:
public_csv_url: "https://opengeofeed.org/feed/public.csv"
nro:
delegated_stats_url: "https://ftp.ripe.net/pub/stats/ripencc/nro-stats/latest/nro-delegated-stats"

14
backend/app/core/datasource_defaults.py
View File

@@ -141,6 +141,20 @@ DEFAULT_DATASOURCES = {
"priority": "P1",
"frequency_minutes": 1440,
},
"opengeofeed_prefix_geo": {
"id": 24,
"name": "OpenGeoFeed Prefix Geography",
"module": "L3",
"priority": "P1",
"frequency_minutes": 1440,
},
"nro_delegated_prefix_geo": {
"id": 25,
"name": "NRO Delegated Prefix Geography",
"module": "L3",
"priority": "P1",
"frequency_minutes": 1440,
},
}
ID_TO_COLLECTOR = {info["id"]: name for name, info in DEFAULT_DATASOURCES.items()}

123
backend/app/services/bgp_enrichment.py
View File

@@ -101,6 +101,49 @@ async def _lookup_prefix_geography(
db: AsyncSession,
prefix_values: list[str],
) -> dict[str, dict[str, Any]]:
async def _query_prefix_metadata(
*,
source: str,
family: str,
range_start: str,
range_end: str,
) -> dict[str, Any] | None:
result = await db.execute(
text(
"""
SELECT metadata
FROM collected_data
WHERE source = :source
AND COALESCE(is_current, TRUE) = TRUE
AND metadata->>'family' = :family
AND CAST(metadata->>'range_start' AS inet) <= CAST(:range_start AS inet)
AND CAST(metadata->>'range_end' AS inet) >= CAST(:range_end AS inet)
ORDER BY
masklen(CAST(metadata->>'prefix' AS cidr)) DESC NULLS LAST,
id DESC
LIMIT 1
"""
),
{
"source": source,
"family": family,
"range_start": range_start,
"range_end": range_end,
},
)
row = result.fetchone()
if not row:
return None
if isinstance(row, dict):
payload = row.get("metadata") or row.get("extra_data")
elif hasattr(row, "_mapping"):
payload = row._mapping.get("metadata") or row._mapping.get("extra_data")
else:
payload = row[0]
return payload if isinstance(payload, dict) else None
results: dict[str, dict[str, Any]] = {}
for prefix in prefix_values:
@@ -112,50 +155,44 @@ async def _lookup_prefix_geography(
family = f"ipv{network.version}"
range_start = str(network.network_address)
range_end = str(network.broadcast_address)
result = await db.execute(
text(
"""
SELECT metadata
FROM collected_data
WHERE source = 'iptoasn_prefix_geo'
AND COALESCE(is_current, TRUE) = TRUE
AND metadata->>'family' = :family
AND CAST(metadata->>'range_start' AS inet) <= CAST(:range_start AS inet)
AND CAST(metadata->>'range_end' AS inet) >= CAST(:range_end AS inet)
ORDER BY id DESC
LIMIT 1
"""
),
{
"family": family,
"range_start": range_start,
"range_end": range_end,
},
payload = await _query_prefix_metadata(
source="opengeofeed_prefix_geo",
family=family,
range_start=range_start,
range_end=range_end,
)
row = result.fetchone()
if not row:
continue
if isinstance(row, dict):
payload = row.get("metadata") or row.get("extra_data")
elif hasattr(row, "_mapping"):
payload = row._mapping.get("metadata") or row._mapping.get("extra_data")
else:
payload = row[0]
if not isinstance(payload, dict):
selected_source = "opengeofeed"
if not payload:
payload = await _query_prefix_metadata(
source="iptoasn_prefix_geo",
family=family,
range_start=range_start,
range_end=range_end,
)
selected_source = "iptoasn"
if not payload:
payload = await _query_prefix_metadata(
source="nro_delegated_prefix_geo",
family=family,
range_start=range_start,
range_end=range_end,
)
selected_source = "nro_delegated"
if not payload:
continue
country = normalize_country(payload.get("country") or payload.get("country_code"))
prefix_hint = payload.get("prefix") or prefix
asn = _safe_int(payload.get("asn"))
as_name = payload.get("as_name")
city = payload.get("city")
centroid = get_country_centroid(country)
regions = []
if country:
regions.append(
{
"country": country,
"city": None,
"city": city,
"latitude": centroid.get("latitude") if centroid else None,
"longitude": centroid.get("longitude") if centroid else None,
}
@@ -164,10 +201,30 @@ async def _lookup_prefix_geography(
results[prefix] = {
"prefix": prefix_hint,
"country": country,
"city": city,
"asn": asn,
"as_name": as_name,
"source": payload.get("source_dataset") or "iptoasn_combined",
"confidence": "country_range",
"source": payload.get("source_dataset")
or (
"opengeofeed_public"
if selected_source == "opengeofeed"
else (
"iptoasn_combined"
if selected_source == "iptoasn"
else "nro_delegated_stats"
)
),
"confidence": payload.get("confidence")
or (
"geofeed"
if selected_source == "opengeofeed"
else (
"country_range"
if selected_source == "iptoasn"
else "registry_allocated"
)
),
"geography_mode": "prefix_geography",
"regions": regions,
}

4
backend/app/services/collectors/__init__.py
View File

@@ -33,6 +33,8 @@ from app.services.collectors.celestrak import CelesTrakTLECollector
from app.services.collectors.ris_live import RISLiveCollector
from app.services.collectors.bgpstream import BGPStreamBackfillCollector
from app.services.collectors.iptoasn import IPtoASNPrefixGeoCollector
from app.services.collectors.opengeofeed import OpenGeoFeedPrefixGeoCollector
from app.services.collectors.nro_delegated import NRODelegatedPrefixGeoCollector
collector_registry.register(TOP500Collector())
collector_registry.register(EpochAIGPUCollector())
@@ -57,3 +59,5 @@ collector_registry.register(CelesTrakTLECollector())
collector_registry.register(RISLiveCollector())
collector_registry.register(BGPStreamBackfillCollector())
collector_registry.register(IPtoASNPrefixGeoCollector())
collector_registry.register(OpenGeoFeedPrefixGeoCollector())
collector_registry.register(NRODelegatedPrefixGeoCollector())

290
backend/app/services/collectors/iptoasn.py
View File

@@ -6,9 +6,15 @@ prefix-to-country/ASN geography hints for BGP enrichment.
from __future__ import annotations
import asyncio
import gzip
import hashlib
import json
import tempfile
import time
from datetime import UTC, datetime
from ipaddress import summarize_address_range, ip_address
from pathlib import Path
from typing import Any
import httpx
@@ -23,40 +29,274 @@ class IPtoASNPrefixGeoCollector(BaseCollector):
frequency_hours = 24
data_type = "prefix_geography"
fail_on_empty = True
_cache_dir = Path(tempfile.gettempdir()) / "planet-download-cache" / "iptoasn"
@staticmethod
def _build_dataset_urls(resolved_url: str) -> list[str]:
if "ip2asn-combined.tsv.gz" in resolved_url:
return [
resolved_url.replace("ip2asn-combined.tsv.gz", "ip2asn-v4.tsv.gz"),
resolved_url.replace("ip2asn-combined.tsv.gz", "ip2asn-v6.tsv.gz"),
]
return [resolved_url]
@staticmethod
def _cache_key(url: str) -> str:
return hashlib.sha1(url.encode("utf-8")).hexdigest()[:16]
@classmethod
def _cache_paths(cls, url: str) -> tuple[Path, Path, Path]:
key = cls._cache_key(url)
gz_path = cls._cache_dir / f"{key}.tsv.gz"
part_path = cls._cache_dir / f"{key}.tsv.gz.part"
meta_path = cls._cache_dir / f"{key}.meta.json"
return gz_path, part_path, meta_path
@staticmethod
def _load_meta(meta_path: Path) -> dict[str, Any]:
if not meta_path.exists():
return {}
try:
return json.loads(meta_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {}
@staticmethod
def _save_meta(meta_path: Path, payload: dict[str, Any]) -> None:
meta_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
@staticmethod
def _validators_match(meta: dict[str, Any], remote: dict[str, Any]) -> bool:
etag = str(remote.get("etag") or "").strip()
last_modified = str(remote.get("last_modified") or "").strip()
if etag:
return etag == str(meta.get("etag") or "").strip()
if last_modified:
return last_modified == str(meta.get("last_modified") or "").strip()
return True
async def _fetch_remote_info(self, client: httpx.AsyncClient, url: str) -> dict[str, Any]:
try:
response = await client.head(url)
if response.status_code >= 400:
return {}
content_length_raw = response.headers.get("content-length")
content_length = int(content_length_raw) if content_length_raw else None
return {
"etag": response.headers.get("etag"),
"last_modified": response.headers.get("last-modified"),
"content_length": content_length,
"accept_ranges": (response.headers.get("accept-ranges") or "").lower(),
}
except (httpx.HTTPError, ValueError):
return {}
def _parse_rows_from_gzip_file(self, file_path: Path) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
with gzip.open(file_path, "rt", encoding="utf-8", errors="replace") as f:
for raw_line in f:
line = raw_line.strip()
if not line or line.startswith("#"):
continue
parts = line.split("\t")
if len(parts) < 5:
continue
range_start, range_end, asn, country_code, as_name = parts[:5]
rows.append(
{
"range_start": range_start,
"range_end": range_end,
"asn": asn,
"country_code": country_code,
"as_name": as_name,
}
)
if not rows:
raise RuntimeError(f"IPtoASN dataset parsed empty rows: {file_path.name}")
return rows
async def _download_dataset_with_resume(
self,
client: httpx.AsyncClient,
url: str,
*,
progress_callback=None,
) -> Path:
self._cache_dir.mkdir(parents=True, exist_ok=True)
gz_path, part_path, meta_path = self._cache_paths(url)
meta = self._load_meta(meta_path)
remote = await self._fetch_remote_info(client, url)
# Reuse existing complete file when validators and size still match.
if gz_path.exists():
local_size = gz_path.stat().st_size
remote_size = remote.get("content_length")
if (
self._validators_match(meta, remote)
and (remote_size is None or local_size == remote_size)
):
try:
self._parse_rows_from_gzip_file(gz_path)
if progress_callback and local_size > 0:
await progress_callback(local_size)
return gz_path
except Exception:
gz_path.unlink(missing_ok=True)
expected_size = remote.get("content_length")
can_resume = (remote.get("accept_ranges") or "") == "bytes"
resume_from = part_path.stat().st_size if part_path.exists() else 0
if expected_size is not None and resume_from > expected_size:
part_path.unlink(missing_ok=True)
resume_from = 0
if not self._validators_match(meta, remote):
part_path.unlink(missing_ok=True)
resume_from = 0
headers = {
"User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)",
"Accept": "application/gzip,application/octet-stream,*/*",
}
if can_resume and resume_from > 0:
headers["Range"] = f"bytes={resume_from}-"
if remote.get("etag"):
headers["If-Range"] = str(remote.get("etag"))
elif remote.get("last_modified"):
headers["If-Range"] = str(remote.get("last_modified"))
async with client.stream("GET", url, headers=headers) as response:
response.raise_for_status()
if response.status_code == 206 and resume_from > 0:
mode = "ab"
else:
mode = "wb"
resume_from = 0
with part_path.open(mode) as f:
downloaded = resume_from
if progress_callback and downloaded > 0:
await progress_callback(downloaded)
async for chunk in response.aiter_bytes():
if chunk:
f.write(chunk)
downloaded += len(chunk)
if progress_callback:
await progress_callback(downloaded)
part_size = part_path.stat().st_size if part_path.exists() else 0
if expected_size is not None and part_size != expected_size:
raise RuntimeError(
f"IPtoASN download incomplete for {url}: expected={expected_size}, got={part_size}"
)
part_path.replace(gz_path)
self._save_meta(
meta_path,
{
"url": url,
"etag": remote.get("etag"),
"last_modified": remote.get("last_modified"),
"content_length": expected_size,
"updated_at": datetime.now(UTC).isoformat(),
},
)
return gz_path
async def _fetch_dataset_rows(
self,
client: httpx.AsyncClient,
url: str,
*,
progress_callback=None,
) -> list[dict[str, Any]]:
try:
gz_path = await self._download_dataset_with_resume(
client,
url,
progress_callback=progress_callback,
)
return self._parse_rows_from_gzip_file(gz_path)
except Exception:
# Bad partial/complete cache or interrupted stream: clear local files
# and retry once from scratch.
gz_path, part_path, meta_path = self._cache_paths(url)
gz_path.unlink(missing_ok=True)
part_path.unlink(missing_ok=True)
meta_path.unlink(missing_ok=True)
gz_path = await self._download_dataset_with_resume(
client,
url,
progress_callback=progress_callback,
)
return self._parse_rows_from_gzip_file(gz_path)
async def fetch(self) -> list[dict[str, Any]]:
if not self._resolved_url:
raise RuntimeError("IPtoASN combined URL is not configured")
dataset_urls = self._build_dataset_urls(self._resolved_url)
async with httpx.AsyncClient(timeout=180.0, follow_redirects=True) as client:
response = await client.get(
self._resolved_url,
headers={
"User-Agent": "Planet-Intelligence-System/1.0 (Python/collector)",
"Accept": "application/gzip,application/octet-stream,*/*",
},
remote_infos = await asyncio.gather(
*(self._fetch_remote_info(client, url) for url in dataset_urls)
)
response.raise_for_status()
body = gzip.decompress(response.content).decode("utf-8", errors="replace")
expected_sizes = [
info.get("content_length")
for info in remote_infos
if isinstance(info.get("content_length"), int)
]
total_expected = sum(expected_sizes) if expected_sizes else 0
if total_expected > 0 and self._current_task and self._db_session:
self._current_task.total_records = total_expected
self._current_task.records_processed = 0
self._current_task.progress = 0.0
await self._db_session.commit()
await self._publish_task_update(force=True)
url_progress: dict[str, int] = {url: 0 for url in dataset_urls}
progress_lock = asyncio.Lock()
last_emit = {"t": 0.0, "value": 0}
min_emit_bytes = max(total_expected // 200, 2 * 1024 * 1024) if total_expected > 0 else 4 * 1024 * 1024
async def on_url_progress(url: str, downloaded_bytes: int) -> None:
if total_expected <= 0:
return
async with progress_lock:
current = max(0, downloaded_bytes)
if current < url_progress[url]:
return
url_progress[url] = current
aggregated = sum(url_progress.values())
now = time.monotonic()
should_emit = (
aggregated >= total_expected
or aggregated - last_emit["value"] >= min_emit_bytes
or now - last_emit["t"] >= 2.0
)
if not should_emit:
return
last_emit["value"] = aggregated
last_emit["t"] = now
await self.update_progress(min(aggregated, total_expected), commit=True)
batches = await asyncio.gather(
*(
self._fetch_dataset_rows(
client,
url,
progress_callback=lambda downloaded, u=url: on_url_progress(u, downloaded),
)
for url in dataset_urls
)
)
if total_expected > 0:
await self.update_progress(total_expected, commit=True, force=True)
rows: list[dict[str, Any]] = []
for raw_line in body.splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
parts = line.split("\t")
if len(parts) < 5:
continue
range_start, range_end, asn, country_code, as_name = parts[:5]
rows.append(
{
"range_start": range_start,
"range_end": range_end,
"asn": asn,
"country_code": country_code,
"as_name": as_name,
}
)
for batch in batches:
rows.extend(batch)
return rows
def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]:

49
backend/app/services/scheduler.py
View File

@@ -18,6 +18,7 @@ from app.services.collectors.registry import collector_registry
logger = logging.getLogger(__name__)
scheduler = AsyncIOScheduler()
RUNNING_TASK_GUARD_TIMEOUT_MINUTES = 90
async def _update_next_run_at(datasource: DataSource, session) -> None:
@@ -76,6 +77,54 @@ async def run_collector_task(collector_name: str):
logger.info("Skipping disabled collector: %s", collector_name)
return
running_result = await db.execute(
select(CollectionTask)
.where(
CollectionTask.datasource_id == datasource.id,
CollectionTask.status == "running",
)
.order_by(CollectionTask.started_at.desc(), CollectionTask.id.desc())
.limit(1)
)
existing_running = running_result.scalar_one_or_none()
if existing_running is not None:
now = datetime.now(UTC)
started_at = existing_running.started_at
if started_at is not None and started_at.tzinfo is None:
started_at = started_at.replace(tzinfo=UTC)
is_stale = (
started_at is not None
and (now - started_at) > timedelta(minutes=RUNNING_TASK_GUARD_TIMEOUT_MINUTES)
)
if not is_stale:
logger.warning(
"Skipping collector %s trigger because task %s is already running",
collector_name,
existing_running.id,
)
return
existing_error = (existing_running.error_message or "").strip()
stale_reason = (
f"Marked failed automatically after stale running timeout "
f"({RUNNING_TASK_GUARD_TIMEOUT_MINUTES}m) in scheduler guard"
)
existing_running.status = "failed"
existing_running.phase = "failed"
existing_running.completed_at = now
existing_running.error_message = (
f"{existing_error}\n{stale_reason}".strip()
if existing_error
else stale_reason
)
await db.commit()
logger.warning(
"Marked stale running task %s as failed before rerun of %s",
existing_running.id,
collector_name,
)
try:
collector._datasource_id = datasource.id
logger.info("Running collector: %s (datasource_id=%s)", collector_name, datasource.id)