fix: recover bgp anomaly and incident generation

backend/app/services/bgp_detectors.py

@@ -9,6 +9,52 @@ from typing import Any
 from app.models.bgp_anomaly import BGPAnomaly
 
 
+def _iter_event_regions(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    regions: list[dict[str, Any]] = []
+    seen: set[tuple[Any, ...]] = set()
+    for event in events:
+        metadata = event.get("metadata") or {}
+        location = metadata.get("collector_location") or {}
+        region = {
+            "collector": metadata.get("collector"),
+            "country": location.get("country"),
+            "city": location.get("city"),
+            "latitude": location.get("latitude"),
+            "longitude": location.get("longitude"),
+        }
+        region_key = (
+            region.get("collector"),
+            region.get("country"),
+            region.get("city"),
+            region.get("latitude"),
+            region.get("longitude"),
+        )
+        if region_key in seen:
+            continue
+        seen.add(region_key)
+        regions.append(region)
+    return regions
+
+
+def _unique_collectors(events: list[dict[str, Any]]) -> list[str]:
+    return sorted(
+        {
+            str((event.get("metadata") or {}).get("collector"))
+            for event in events
+            if (event.get("metadata") or {}).get("collector")
+        }
+    )
+
+
+def _unique_peers(events: list[dict[str, Any]]) -> list[int]:
+    peers: set[int] = set()
+    for event in events:
+        peer_asn = (event.get("metadata") or {}).get("peer_asn")
+        if peer_asn is not None:
+            peers.add(int(peer_asn))
+    return sorted(peers)
+
+
 def detect_origin_change_anomalies(
     *,
     source: str,
@@ -29,14 +75,24 @@ def detect_origin_change_anomalies(
     for prefix, origins in prefix_to_origins.items():
         historic = previous_origin_map.get(prefix, set())
         new_origins = sorted(origin for origin in origins if origin not in historic)
-        if not historic or not new_origins:
+        related_events = [
+            event
+            for event in events
+            if (event.get("metadata") or {}).get("prefix") == prefix
+        ]
+        related_collectors = _unique_collectors(related_events)
+        related_regions = _iter_event_regions(related_events)
+
+        moas_candidate = not historic and len(origins) >= 2 and len(related_collectors) >= 2
+        if (not historic or not new_origins) and not moas_candidate:
             continue
 
-        for new_origin in new_origins:
+        target_origins = new_origins or sorted(origins)
+        for new_origin in target_origins:
             sample_event = next(
                 (
                     event
-                    for event in events
+                    for event in related_events
                     if (event.get("metadata") or {}).get("prefix") == prefix
                     and int((event.get("metadata") or {}).get("origin_asn") or -1) == new_origin
                 ),
@@ -44,31 +100,49 @@ def detect_origin_change_anomalies(
             )
             sample_metadata = sample_event.get("metadata") or {}
             sample_enrichment = sample_metadata.get("enrichment") or {}
+            anomaly_type = "origin_change"
+            severity = "critical"
+            confidence = 0.86
+            summary = f"Prefix {prefix} is now originated by AS{new_origin}, outside the current baseline."
+            evidence_previous_origins = sorted(historic)
+            if moas_candidate and not historic:
+                anomaly_type = "origin_conflict"
+                severity = "high"
+                confidence = 0.74
+                summary = (
+                    f"Prefix {prefix} is being originated by multiple ASNs "
+                    f"{sorted(origins)} across {len(related_collectors)} collectors."
+                )
+                evidence_previous_origins = []
             anomalies.append(
                 BGPAnomaly(
                     snapshot_id=snapshot_id,
                     task_id=task_id,
                     source=source,
-                    anomaly_type="origin_change",
-                    severity="critical",
+                    anomaly_type=anomaly_type,
+                    severity=severity,
                     status="active",
-                    entity_key=f"origin_change:{prefix}:{new_origin}",
+                    entity_key=f"{anomaly_type}:{prefix}:{new_origin}",
                     prefix=prefix,
-                    origin_asn=sorted(historic)[0],
+                    origin_asn=sorted(historic)[0] if historic else None,
                     new_origin_asn=new_origin,
-                    peer_scope=[],
+                    peer_scope=related_collectors,
                     started_at=datetime.now(UTC),
-                    confidence=0.86,
-                    summary=f"Prefix {prefix} is now originated by AS{new_origin}, outside the current baseline.",
+                    confidence=confidence,
+                    summary=summary,
                     evidence={
-                        "previous_origins": sorted(historic),
+                        "previous_origins": evidence_previous_origins,
                         "current_origins": sorted(origins),
-                        "events": [sample_metadata] if sample_metadata else [],
+                        "events": [
+                            (item.get("metadata") or {})
+                            for item in related_events[:10]
+                        ],
                         "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
                         "new_origin_asn_profile": sample_enrichment.get("new_origin_asn_profile"),
                         "rpki_validation": sample_enrichment.get("rpki_validation"),
                         "prefix_scope": sample_enrichment.get("prefix_scope"),
-                        "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                        "impacted_regions": related_regions
+                        or sample_enrichment.get("prefix_scope", {}).get("regions", []),
                     },
                 )
             )
@@ -86,18 +160,27 @@ def detect_more_specific_burst_anomalies(
     prefix_to_more_specifics: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
     for event in events:
         metadata = event.get("metadata") or {}
-        prefix = metadata.get("prefix")
         enrichment = metadata.get("enrichment") or {}
-        if prefix and enrichment.get("is_more_specific"):
-            prefix_to_more_specifics[str(prefix).split("/")[0]].append(event)
+        root_prefix = enrichment.get("prefix_supernet")
+        if root_prefix and enrichment.get("is_more_specific"):
+            prefix_to_more_specifics[str(root_prefix)].append(event)
 
     anomalies: list[BGPAnomaly] = []
     for root_prefix, more_specifics in prefix_to_more_specifics.items():
-        if len(more_specifics) < 2:
+        unique_prefixes = sorted(
+            {
+                str((item.get("metadata") or {}).get("prefix"))
+                for item in more_specifics
+                if (item.get("metadata") or {}).get("prefix")
+            }
+        )
+        related_collectors = _unique_collectors(more_specifics)
+        if len(unique_prefixes) < 2 and len(related_collectors) < 2:
             continue
 
         sample = more_specifics[0].get("metadata") or {}
         sample_enrichment = sample.get("enrichment") or {}
+        event_count = len(more_specifics)
         anomalies.append(
             BGPAnomaly(
                 snapshot_id=snapshot_id,
@@ -106,26 +189,25 @@ def detect_more_specific_burst_anomalies(
                 anomaly_type="more_specific_burst",
                 severity="high",
                 status="active",
-                entity_key=f"more_specific_burst:{root_prefix}:{len(more_specifics)}",
+                entity_key=f"more_specific_burst:{root_prefix}:{len(unique_prefixes)}:{len(related_collectors)}",
                 prefix=sample.get("prefix"),
                 origin_asn=sample.get("origin_asn"),
                 new_origin_asn=None,
-                peer_scope=sorted(
-                    {
-                        str(item.get("metadata", {}).get("collector") or "")
-                        for item in more_specifics
-                        if item.get("metadata", {}).get("collector")
-                    }
-                ),
+                peer_scope=related_collectors,
                 started_at=datetime.now(UTC),
-                confidence=0.72,
-                summary=f"{len(more_specifics)} more-specific announcements clustered around {root_prefix}.",
+                confidence=min(0.64 + (0.04 * min(event_count, 5)), 0.88),
+                summary=(
+                    f"{len(unique_prefixes)} more-specific prefixes clustered under {root_prefix} "
+                    f"across {len(related_collectors) or 1} collectors."
+                ),
                 evidence={
                     "events": [item.get("metadata") for item in more_specifics[:10]],
+                    "unique_prefixes": unique_prefixes,
                     "rpki_validation": sample_enrichment.get("rpki_validation"),
                     "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
                     "prefix_scope": sample_enrichment.get("prefix_scope"),
-                    "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                    "impacted_regions": _iter_event_regions(more_specifics)
+                    or sample_enrichment.get("prefix_scope", {}).get("regions", []),
                 },
             )
         )
@@ -141,27 +223,30 @@ def detect_mass_withdrawal_anomalies(
     events: list[dict[str, Any]],
 ) -> list[BGPAnomaly]:
     withdrawal_counter: Counter[tuple[str, int | None]] = Counter()
+    withdrawal_events_by_key: defaultdict[tuple[str, int | None], list[dict[str, Any]]] = defaultdict(list)
     for event in events:
         metadata = event.get("metadata") or {}
         prefix = metadata.get("prefix")
         if prefix and metadata.get("event_type") == "withdrawal":
-            withdrawal_counter[(str(prefix), metadata.get("origin_asn"))] += 1
+            key = (str(prefix), metadata.get("origin_asn"))
+            withdrawal_counter[key] += 1
+            withdrawal_events_by_key[key].append(event)
 
     anomalies: list[BGPAnomaly] = []
     for (prefix, origin_asn), count in withdrawal_counter.items():
-        if count < 3:
+        related_events = withdrawal_events_by_key[(prefix, origin_asn)]
+        related_collectors = _unique_collectors(related_events)
+        related_peers = _unique_peers(related_events)
+        if count < 3 and not (count >= 2 and len(related_collectors) >= 2):
             continue
-        sample_event = next(
-            (
-                event
-                for event in events
-                if (event.get("metadata") or {}).get("prefix") == prefix
-                and (event.get("metadata") or {}).get("event_type") == "withdrawal"
-            ),
-            {},
-        )
+        sample_event = related_events[0] if related_events else {}
         sample_metadata = sample_event.get("metadata") or {}
         sample_enrichment = sample_metadata.get("enrichment") or {}
+        severity = "medium"
+        if count >= 4 or len(related_collectors) >= 3:
+            severity = "high"
+        if count >= 8:
+            severity = "critical"
 
         anomalies.append(
             BGPAnomaly(
@@ -169,23 +254,32 @@ def detect_mass_withdrawal_anomalies(
                 task_id=task_id,
                 source=source,
                 anomaly_type="mass_withdrawal",
-                severity="high" if count < 8 else "critical",
+                severity=severity,
                 status="active",
-                entity_key=f"mass_withdrawal:{prefix}:{origin_asn}:{count}",
+                entity_key=f"mass_withdrawal:{prefix}:{origin_asn}:{len(related_collectors)}:{count}",
                 prefix=prefix,
                 origin_asn=origin_asn,
                 new_origin_asn=None,
-                peer_scope=[],
+                peer_scope=related_collectors,
                 started_at=datetime.now(UTC),
-                confidence=min(0.55 + (count * 0.05), 0.95),
-                summary=f"{count} withdrawal events observed for {prefix} in the current ingest window.",
+                confidence=min(0.5 + (count * 0.06) + (0.04 * max(len(related_collectors) - 1, 0)), 0.95),
+                summary=(
+                    f"{count} withdrawal events observed for {prefix} "
+                    f"across {len(related_collectors) or 1} collectors in the current ingest window."
+                ),
                 evidence={
                     "withdrawal_count": count,
-                    "events": [sample_metadata] if sample_metadata else [],
+                    "collector_count": len(related_collectors),
+                    "peer_count": len(related_peers),
+                    "events": [
+                        (item.get("metadata") or {})
+                        for item in related_events[:10]
+                    ],
                     "origin_asn_profile": sample_enrichment.get("origin_asn_profile"),
                     "rpki_validation": sample_enrichment.get("rpki_validation"),
                     "prefix_scope": sample_enrichment.get("prefix_scope"),
-                    "impacted_regions": sample_enrichment.get("prefix_scope", {}).get("regions", []),
+                    "impacted_regions": _iter_event_regions(related_events)
+                    or sample_enrichment.get("prefix_scope", {}).get("regions", []),
                 },
             )
         )

backend/app/services/collectors/bgp_common.py

@@ -293,6 +293,12 @@ async def create_bgp_anomalies_for_batch(
         )
     )
     existing_keys = {row[0] for row in existing_result.fetchall()}
+    existing_anomalies: list[BGPAnomaly] = []
+    if existing_keys:
+        existing_anomaly_result = await db.execute(
+            select(BGPAnomaly).where(BGPAnomaly.entity_key.in_(sorted(existing_keys)))
+        )
+        existing_anomalies = existing_anomaly_result.scalars().all()
 
     created = 0
     created_anomalies: list[BGPAnomaly] = []
@@ -305,11 +311,13 @@ async def create_bgp_anomalies_for_batch(
 
     if created:
         await db.commit()
+    incident_seed_anomalies = [*created_anomalies, *existing_anomalies]
+    if incident_seed_anomalies:
         await create_bgp_incidents_for_anomalies(
             db,
             source=source,
             snapshot_id=snapshot_id,
             task_id=task_id,
-            anomalies=created_anomalies,
+            anomalies=incident_seed_anomalies,
         )
     return created