From 016507ad6845d1aff1d33570621b65adc66d3a12 Mon Sep 17 00:00:00 2001 From: linkong Date: Tue, 31 Mar 2026 18:17:55 +0800 Subject: [PATCH] fix: recover bgp anomaly and incident generation --- VERSION | 2 +- backend/app/services/bgp_detectors.py | 186 +++++++++++++----- backend/app/services/collectors/bgp_common.py | 10 +- backend/tests/test_bgp.py | 117 ++++++++++- docs/CHANGELOG.md | 24 +++ frontend/package-lock.json | 4 +- frontend/package.json | 2 +- pyproject.toml | 2 +- 8 files changed, 289 insertions(+), 58 deletions(-) diff --git a/VERSION b/VERSION index 4240544f..03035cdd 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.22.4 +0.22.5 diff --git a/backend/app/services/bgp_detectors.py b/backend/app/services/bgp_detectors.py index 0f524814..39f7e39d 100644 --- a/backend/app/services/bgp_detectors.py +++ b/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", []), }, ) ) diff --git a/backend/app/services/collectors/bgp_common.py b/backend/app/services/collectors/bgp_common.py index 2ae0e38a..9625aa40 100644 --- a/backend/app/services/collectors/bgp_common.py +++ b/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 diff --git a/backend/tests/test_bgp.py b/backend/tests/test_bgp.py index 57a483a8..a84f28de 100644 --- a/backend/tests/test_bgp.py +++ b/backend/tests/test_bgp.py @@ -1,6 +1,6 @@ """Tests for BGP observability helpers.""" -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest from httpx import ASGITransport, AsyncClient @@ -10,7 +10,10 @@ from app.api.v1.bgp import BGP_SOURCES from app.core.security import get_current_user from app.db.session import get_db from app.main import app -from app.services.bgp_detectors import detect_mass_withdrawal_anomalies +from app.services.bgp_detectors import ( + detect_mass_withdrawal_anomalies, + detect_origin_change_anomalies, +) from app.services.collectors.bgp_common import ( create_bgp_anomalies_for_batch, save_bgp_observations_for_batch, @@ -223,6 +226,95 @@ def test_detect_mass_withdrawal_anomalies(): assert anomalies[0].prefix == "203.0.113.0/24" +def test_detect_origin_change_anomalies_creates_conflict_without_baseline(): + events = [ + { + "metadata": { + "prefix": "203.0.113.0/24", + "origin_asn": 64496, + "collector": "rrc00", + "collector_location": { + "country": "Netherlands", + "city": "Amsterdam", + "latitude": 52.3676, + "longitude": 4.9041, + }, + } + }, + { + "metadata": { + "prefix": "203.0.113.0/24", + "origin_asn": 64497, + "collector": "rrc01", + "collector_location": { + "country": "United Kingdom", + "city": "London", + "latitude": 51.5072, + "longitude": -0.1276, + }, + } + }, + ] + + anomalies = detect_origin_change_anomalies( + source="ris_live_bgp", + snapshot_id=1, + task_id=2, + events=events, + previous_origin_map={}, + ) + + assert len(anomalies) == 2 + assert {item.anomaly_type for item in anomalies} == {"origin_conflict"} + assert anomalies[0].peer_scope == ["rrc00", "rrc01"] + + +def test_detect_mass_withdrawal_anomalies_accepts_cross_collector_pair(): + events = [ + { + "metadata": { + "prefix": "203.0.113.0/24", + "origin_asn": 64496, + "event_type": "withdrawal", + "collector": "rrc00", + "peer_asn": 3333, + "collector_location": { + "country": "Netherlands", + "city": "Amsterdam", + "latitude": 52.3676, + "longitude": 4.9041, + }, + } + }, + { + "metadata": { + "prefix": "203.0.113.0/24", + "origin_asn": 64496, + "event_type": "withdrawal", + "collector": "rrc01", + "peer_asn": 3334, + "collector_location": { + "country": "United Kingdom", + "city": "London", + "latitude": 51.5072, + "longitude": -0.1276, + }, + } + }, + ] + + anomalies = detect_mass_withdrawal_anomalies( + source="ris_live_bgp", + snapshot_id=1, + task_id=2, + events=events, + ) + + assert len(anomalies) == 1 + assert anomalies[0].severity == "medium" + assert anomalies[0].evidence["collector_count"] == 2 + + def test_bgp_incident_to_dict(): incident = BGPIncident( source="ris_live_bgp", @@ -405,6 +497,7 @@ async def test_infer_related_infrastructure_links_nearby_cables(): @pytest.mark.asyncio async def test_build_bgp_collector_coverage_summarizes_observations(): + now = datetime.now(UTC) obs_one = BGPObservation( source="ris_live_bgp", collector="rrc00", @@ -412,7 +505,7 @@ async def test_build_bgp_collector_coverage_summarizes_observations(): origin_asn=64496, peer_asn=3333, event_type="announcement", - observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + observed_at=now, collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) obs_two = BGPObservation( @@ -422,7 +515,7 @@ async def test_build_bgp_collector_coverage_summarizes_observations(): origin_asn=64497, peer_asn=3334, event_type="withdrawal", - observed_at=datetime(2026, 3, 30, 10, 5, tzinfo=UTC), + observed_at=now + timedelta(minutes=5), collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) db = _FakeAsyncSession([[obs_one, obs_two]]) @@ -543,11 +636,22 @@ async def test_create_bgp_anomalies_for_batch_skips_existing_entity_keys(): extra_data={"prefix": "203.0.113.0/24", "origin_asn": 64496}, ) existing_key = ("origin_change:203.0.113.0/24:64497",) + existing_anomaly = BGPAnomaly( + source="ris_live_bgp", + anomaly_type="origin_change", + severity="critical", + status="active", + entity_key="origin_change:203.0.113.0/24:64497", + prefix="203.0.113.0/24", + origin_asn=64496, + new_origin_asn=64497, + ) db = _FakeAsyncSession([ [], [], [previous_record], [existing_key], + [existing_anomaly], ]) events = [ { @@ -578,7 +682,7 @@ async def test_create_bgp_anomalies_for_batch_skips_existing_entity_keys(): assert created == 0 assert len(db.added) == 0 - incident_mock.assert_not_awaited() + incident_mock.assert_awaited_once() async def _bgp_test_client(db_session): @@ -740,6 +844,7 @@ async def test_bgp_event_summary_api_returns_aggregates(): @pytest.mark.asyncio async def test_bgp_collectors_api_returns_coverage(): + now = datetime.now(UTC) observation = BGPObservation( id=1, source="ris_live_bgp", @@ -748,7 +853,7 @@ async def test_bgp_collectors_api_returns_coverage(): prefix="203.0.113.0/24", event_type="announcement", origin_asn=64496, - observed_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + observed_at=now, collector_geo={"city": "Amsterdam", "country": "Netherlands"}, ) db = _FakeAsyncSession([[observation], [observation]]) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index dab9b1fa..5750884b 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,30 @@ This project follows the repository versioning rule: - `feature` -> `+0.1.0` - `bugfix` -> `+0.0.1` +## 0.22.5 + +Released: 2026-03-31 + +### Highlights + +- Relaxed the BGP anomaly pipeline so realtime observation batches can produce visible anomaly and incident signals more consistently instead of staying observation-only. +- Added incident backfill-on-detection behavior so Earth and the BGP console can recover incident objects even when matching anomaly rows already existed from earlier ingests. +- Tightened BGP detector test coverage around low-signal withdrawal bursts, origin conflicts without historic baseline, and anomaly-to-incident regeneration. + +### Improved + +- Improved [bgp_detectors.py](/home/ray/dev/linkong/planet/backend/app/services/bgp_detectors.py) by broadening origin-change detection into a multi-origin conflict path when multiple collectors observe competing origins without a prior baseline. +- Improved [bgp_detectors.py](/home/ray/dev/linkong/planet/backend/app/services/bgp_detectors.py) so more-specific burst detection groups on normalized supernets and accepts cross-collector clusters instead of only a same-root count heuristic. +- Improved [bgp_detectors.py](/home/ray/dev/linkong/planet/backend/app/services/bgp_detectors.py) so mass-withdrawal detection can trigger on smaller but cross-collector withdrawal pairs, with severity and confidence scaled by collector spread and event count. +- Improved anomaly evidence payloads in [bgp_detectors.py](/home/ray/dev/linkong/planet/backend/app/services/bgp_detectors.py) with collector counts, peer counts, unique prefixes, and deduplicated impacted regions, which gives Earth and downstream incident views stronger context. +- Improved [bgp_common.py](/home/ray/dev/linkong/planet/backend/app/services/collectors/bgp_common.py) so incident aggregation now seeds from both newly created anomalies and already-existing matching anomalies, allowing missing incidents to be rebuilt during later ingests. + +### Fixed + +- Fixed the “observations exist but anomalies/incidents stay at zero” failure mode where realtime BGP batches often produced no visible signals because detector thresholds were too strict for live traffic windows. +- Fixed the “existing anomaly but missing incident” gap where the pipeline only created incidents from freshly inserted anomaly rows and skipped rebuilding incident objects for already-known anomaly keys. +- Fixed stale BGP coverage test expectations in [test_bgp.py](/home/ray/dev/linkong/planet/backend/tests/test_bgp.py) by anchoring recent-window assertions to current UTC time instead of hard-coded past timestamps. + ## 0.22.4 Released: 2026-03-31 diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 0378a31e..c0241678 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "planet-frontend", - "version": "0.22.4", + "version": "0.22.5", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "planet-frontend", - "version": "0.22.4", + "version": "0.22.5", "dependencies": { "@ant-design/icons": "^5.2.6", "antd": "^5.12.5", diff --git a/frontend/package.json b/frontend/package.json index 96edc048..683b4404 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "planet-frontend", - "version": "0.22.4", + "version": "0.22.5", "private": true, "dependencies": { "@ant-design/icons": "^5.2.6", diff --git a/pyproject.toml b/pyproject.toml index 10766016..877c24df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "planet" -version = "0.22.2" +version = "0.22.5" description = "智能星球计划 - 态势感知系统" requires-python = ">=3.14" dependencies = [