132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""RIPE RIS Live collector."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
from app.services.collectors.base import BaseCollector
|
|
from app.services.collectors.bgp_common import create_bgp_anomalies_for_batch, normalize_bgp_event
|
|
|
|
|
|
class RISLiveCollector(BaseCollector):
|
|
name = "ris_live_bgp"
|
|
priority = "P1"
|
|
module = "L3"
|
|
frequency_hours = 1
|
|
data_type = "bgp_update"
|
|
fail_on_empty = True
|
|
max_messages = 100
|
|
idle_timeout_seconds = 15
|
|
|
|
async def fetch(self) -> list[dict[str, Any]]:
|
|
if not self._resolved_url:
|
|
raise RuntimeError("RIS Live URL is not configured")
|
|
|
|
return await asyncio.to_thread(self._fetch_via_stream)
|
|
|
|
def _fetch_via_stream(self) -> list[dict[str, Any]]:
|
|
events: list[dict[str, Any]] = []
|
|
stream_url = "https://ris-live.ripe.net/v1/stream/?format=json&client=planet-ris-live"
|
|
subscribe = json.dumps(
|
|
{
|
|
"host": "rrc00",
|
|
"type": "UPDATE",
|
|
"require": "announcements",
|
|
}
|
|
)
|
|
request = urllib.request.Request(
|
|
stream_url,
|
|
headers={"X-RIS-Subscribe": subscribe},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=20) as response:
|
|
while len(events) < self.max_messages:
|
|
line = response.readline().decode().strip()
|
|
if not line:
|
|
break
|
|
payload = json.loads(line)
|
|
if payload.get("type") != "ris_message":
|
|
continue
|
|
data = payload.get("data", {})
|
|
if isinstance(data, dict):
|
|
events.append(data)
|
|
return events
|
|
|
|
def transform(self, raw_data: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
transformed: list[dict[str, Any]] = []
|
|
for item in raw_data:
|
|
announcements = item.get("announcements") or []
|
|
withdrawals = item.get("withdrawals") or []
|
|
|
|
for announcement in announcements:
|
|
next_hop = announcement.get("next_hop")
|
|
for prefix in announcement.get("prefixes") or []:
|
|
transformed.append(
|
|
normalize_bgp_event(
|
|
{
|
|
**item,
|
|
"collector": item.get("host", "").replace(".ripe.net", ""),
|
|
"event_type": "announcement",
|
|
"prefix": prefix,
|
|
"next_hop": next_hop,
|
|
},
|
|
project="ris-live",
|
|
)
|
|
)
|
|
|
|
for prefix in withdrawals:
|
|
transformed.append(
|
|
normalize_bgp_event(
|
|
{
|
|
**item,
|
|
"collector": item.get("host", "").replace(".ripe.net", ""),
|
|
"event_type": "withdrawal",
|
|
"prefix": prefix,
|
|
},
|
|
project="ris-live",
|
|
)
|
|
)
|
|
|
|
if not announcements and not withdrawals:
|
|
transformed.append(
|
|
normalize_bgp_event(
|
|
{
|
|
**item,
|
|
"collector": item.get("host", "").replace(".ripe.net", ""),
|
|
},
|
|
project="ris-live",
|
|
)
|
|
)
|
|
|
|
self._latest_transformed_batch = transformed
|
|
return transformed
|
|
|
|
async def run(self, db):
|
|
result = await super().run(db)
|
|
if result.get("status") != "success":
|
|
return result
|
|
|
|
snapshot_id = await self._resolve_snapshot_id(db, result.get("task_id"))
|
|
anomaly_count = await create_bgp_anomalies_for_batch(
|
|
db,
|
|
source=self.name,
|
|
snapshot_id=snapshot_id,
|
|
task_id=result.get("task_id"),
|
|
events=getattr(self, "_latest_transformed_batch", []),
|
|
)
|
|
result["anomalies_created"] = anomaly_count
|
|
return result
|
|
|
|
async def _resolve_snapshot_id(self, db, task_id: int | None) -> int | None:
|
|
if task_id is None:
|
|
return None
|
|
from sqlalchemy import select
|
|
from app.models.data_snapshot import DataSnapshot
|
|
|
|
result = await db.execute(
|
|
select(DataSnapshot.id).where(DataSnapshot.task_id == task_id).order_by(DataSnapshot.id.desc())
|
|
)
|
|
return result.scalar_one_or_none()
|