mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
routermon: add WAN geo map + fix batch geo enrichment
- Add WanGeoStatsView (api/wan/geo/) returning country choropleth + bubble data for WAN source IPs using IPGeoCache lat/lon - Add wanGeoMap (420px) in dashboard between ports chart and live table: red colour scale by country, rose bubbles sized by attempt count, top-5 country legend - Rewrite _geo_worker_loop to drain up to 200 queue items per cycle, bulk DB cache check, batch API calls (100 IPs per request) instead of 1 request per IP — fixes ip-api.com timeout pile-up under scan volumes - Add _geo_skip in-memory set: IPs that return no geo data get an empty IPGeoCache placeholder saved to DB so they are never retried; skip set also prevents re-queuing known-empty IPs during flush - Remove _fetch_geo / _enrich_dns_geo / _enrich_wan_geo helpers (logic consolidated into the new batch worker) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Copilot
+113
-61
@@ -235,10 +235,11 @@ def _flush(batch: list):
|
||||
for ev in wan_events
|
||||
]
|
||||
WanEvent.objects.bulk_create(wan_objs)
|
||||
# Enqueue unique source IPs for geo enrichment (non-blocking)
|
||||
# Enqueue unique source IPs for geo enrichment (non-blocking).
|
||||
# Skip IPs already known to have no geo data to reduce queue churn.
|
||||
seen_ips = set()
|
||||
for ev in wan_events:
|
||||
if ev.src_ip not in seen_ips:
|
||||
if ev.src_ip not in seen_ips and ev.src_ip not in _geo_skip:
|
||||
seen_ips.add(ev.src_ip)
|
||||
try:
|
||||
_geo_queue.put_nowait(('wan', ev.src_ip))
|
||||
@@ -289,73 +290,124 @@ def _expire_pending(now):
|
||||
|
||||
# ── Geo enrichment worker ─────────────────────────────────────────────────────
|
||||
|
||||
# In-memory set of IPs that returned no geo data — avoids re-queuing them.
|
||||
# Cleared when it grows too large to prevent unbounded memory use.
|
||||
_geo_skip: set[str] = set()
|
||||
_GEO_SKIP_MAX = 10_000
|
||||
|
||||
|
||||
def _geo_worker_loop():
|
||||
"""Single background thread draining the geo enrichment queue."""
|
||||
"""
|
||||
Batch geo enrichment worker.
|
||||
|
||||
Drains up to 200 queue items per cycle, deduplicates IPs, performs a
|
||||
single bulk DB cache check, then calls the ip-api.com batch endpoint
|
||||
(100 IPs per request) only for IPs that are truly missing from the cache.
|
||||
Failed IPs get an empty IPGeoCache placeholder saved to DB so they are
|
||||
never looked up again.
|
||||
"""
|
||||
logger.info('routermon: geo worker started')
|
||||
|
||||
from nginxmon.geo import _lookup_batch
|
||||
from nginxmon.models import IPGeoCache
|
||||
from routermon.models import DnsQuery, WanEvent
|
||||
|
||||
while True:
|
||||
# Block until at least one item arrives.
|
||||
try:
|
||||
item = _geo_queue.get(timeout=5)
|
||||
first = _geo_queue.get(timeout=5)
|
||||
except queue.Empty:
|
||||
continue
|
||||
kind, value = item
|
||||
|
||||
# Non-blocking drain — collect up to 200 items before processing.
|
||||
batch = [first]
|
||||
while len(batch) < 200:
|
||||
try:
|
||||
batch.append(_geo_queue.get_nowait())
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
dns_pks = [v for k, v in batch if k == 'dns']
|
||||
wan_ips = list({v for k, v in batch if k == 'wan'}) # deduplicated
|
||||
|
||||
# Resolve DNS pks → resolved_ip in one query.
|
||||
dns_pk_ip: dict[int, str] = {}
|
||||
if dns_pks:
|
||||
for row in DnsQuery.objects.filter(
|
||||
pk__in=dns_pks, country=''
|
||||
).values('pk', 'resolved_ip'):
|
||||
if row['resolved_ip']:
|
||||
dns_pk_ip[row['pk']] = row['resolved_ip']
|
||||
|
||||
all_ips: set[str] = set(dns_pk_ip.values()) | set(wan_ips)
|
||||
if not all_ips:
|
||||
continue
|
||||
|
||||
try:
|
||||
if kind == 'dns':
|
||||
_enrich_dns_geo(value)
|
||||
elif kind == 'wan':
|
||||
_enrich_wan_geo(value)
|
||||
# Bulk DB cache check — one query for all IPs.
|
||||
geo_cache: dict[str, IPGeoCache] = {
|
||||
c.ip: c for c in IPGeoCache.objects.filter(ip__in=all_ips)
|
||||
}
|
||||
|
||||
# Only fetch IPs not already in the DB (even empty entries count as "done").
|
||||
missing = [
|
||||
ip for ip in all_ips
|
||||
if ip not in geo_cache and ip not in _geo_skip
|
||||
]
|
||||
|
||||
if missing:
|
||||
for i in range(0, len(missing), 100):
|
||||
chunk = missing[i:i + 100]
|
||||
results = _lookup_batch(chunk)
|
||||
|
||||
# Upsert successful lookups.
|
||||
for ip, data in results.items():
|
||||
obj, _ = IPGeoCache.objects.get_or_create(ip=ip)
|
||||
obj.country = data.get('country', '')
|
||||
obj.country_code = data.get('countryCode', '')
|
||||
obj.region = data.get('regionName', '')
|
||||
obj.city = data.get('city', '')
|
||||
obj.lat = data.get('lat')
|
||||
obj.lon = data.get('lon')
|
||||
obj.isp = data.get('isp', '')
|
||||
obj.save()
|
||||
geo_cache[ip] = obj
|
||||
|
||||
# Save empty placeholder for every IP that failed — prevents
|
||||
# future DB misses and stops the API from being retried.
|
||||
failed = set(chunk) - set(results)
|
||||
if failed:
|
||||
IPGeoCache.objects.bulk_create(
|
||||
[IPGeoCache(ip=ip) for ip in failed],
|
||||
ignore_conflicts=True,
|
||||
)
|
||||
_geo_skip.update(failed)
|
||||
if len(_geo_skip) > _GEO_SKIP_MAX:
|
||||
_geo_skip.clear()
|
||||
|
||||
# Apply geo data to DNS records.
|
||||
for pk, ip in dns_pk_ip.items():
|
||||
geo = geo_cache.get(ip)
|
||||
if geo and geo.country:
|
||||
DnsQuery.objects.filter(pk=pk, country='').update(
|
||||
country=geo.country,
|
||||
country_code=geo.country_code,
|
||||
city=geo.city,
|
||||
)
|
||||
|
||||
# Apply geo data to WAN records (bulk-update per unique IP).
|
||||
for ip in set(wan_ips):
|
||||
geo = geo_cache.get(ip)
|
||||
if geo and geo.country:
|
||||
WanEvent.objects.filter(src_ip=ip, country='').update(
|
||||
country=geo.country,
|
||||
country_code=geo.country_code,
|
||||
city=geo.city,
|
||||
isp=getattr(geo, 'isp', ''),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
logger.debug('routermon: geo enrichment error (%s %s): %s', kind, value, exc)
|
||||
|
||||
|
||||
def _fetch_geo(ip: str):
|
||||
"""Return IPGeoCache object for ip, fetching from ip-api.com if needed."""
|
||||
from nginxmon.models import IPGeoCache
|
||||
obj = IPGeoCache.objects.filter(ip=ip).first()
|
||||
if obj and obj.country:
|
||||
return obj
|
||||
# Fetch from ip-api.com
|
||||
from nginxmon.geo import _lookup_batch
|
||||
results = _lookup_batch([ip])
|
||||
data = results.get(ip, {})
|
||||
if not data:
|
||||
return None
|
||||
obj, _ = IPGeoCache.objects.get_or_create(ip=ip)
|
||||
obj.country = data.get('country', '')
|
||||
obj.country_code = data.get('countryCode', '')
|
||||
obj.region = data.get('regionName', '')
|
||||
obj.city = data.get('city', '')
|
||||
obj.lat = data.get('lat')
|
||||
obj.lon = data.get('lon')
|
||||
obj.isp = data.get('isp', '')
|
||||
obj.save()
|
||||
return obj
|
||||
|
||||
|
||||
def _enrich_dns_geo(pk: int):
|
||||
from routermon.models import DnsQuery
|
||||
row = DnsQuery.objects.filter(pk=pk, country='').values('pk', 'resolved_ip').first()
|
||||
if not row or not row['resolved_ip']:
|
||||
return
|
||||
geo = _fetch_geo(row['resolved_ip'])
|
||||
if geo and geo.country:
|
||||
DnsQuery.objects.filter(pk=pk, country='').update(
|
||||
country=geo.country,
|
||||
country_code=geo.country_code,
|
||||
city=geo.city,
|
||||
)
|
||||
|
||||
|
||||
def _enrich_wan_geo(src_ip: str):
|
||||
"""Geo-enrich all WanEvent rows for src_ip that have no country yet."""
|
||||
from routermon.models import WanEvent
|
||||
geo = _fetch_geo(src_ip)
|
||||
if geo and geo.country:
|
||||
WanEvent.objects.filter(src_ip=src_ip, country='').update(
|
||||
country=geo.country,
|
||||
country_code=geo.country_code,
|
||||
city=geo.city,
|
||||
isp=geo.isp,
|
||||
)
|
||||
logger.warning('routermon: geo batch error: %s', exc)
|
||||
|
||||
|
||||
def _build_excluded_set(text: str) -> set:
|
||||
|
||||
@@ -292,6 +292,15 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- WAN geo map -->
|
||||
<div class="px-4 pt-3 pb-2">
|
||||
<div class="flex items-center justify-between mb-2 flex-wrap gap-2">
|
||||
<div class="text-xs font-medium text-gray-500">Source Geography — {{ range_label }}</div>
|
||||
<div id="wan-geo-legend" class="flex items-center gap-3 text-xs text-gray-400 flex-wrap"></div>
|
||||
</div>
|
||||
<div id="wanGeoMap" class="w-full" style="height:420px; position:relative; overflow:hidden;"></div>
|
||||
</div>
|
||||
|
||||
<!-- Live WAN event table -->
|
||||
<div id="wan-container"
|
||||
class="overflow-x-auto"
|
||||
@@ -516,6 +525,78 @@ wanPortsChart.setOption({
|
||||
loadWanChart();
|
||||
window.addEventListener('resize', () => { wanTimelineChart.resize(); wanPortsChart.resize(); });
|
||||
|
||||
// ── WAN geo map ───────────────────────────────────────────────────────────────
|
||||
async function loadWanGeoMap() {
|
||||
const range = '{{ current_range }}';
|
||||
const [geoRes, worldRes] = await Promise.all([
|
||||
fetch("{% url 'routermon-wan-geo' %}?range=" + range),
|
||||
fetch('https://cdn.jsdelivr.net/npm/world-atlas@2/countries-110m.json'),
|
||||
]);
|
||||
const geo = await geoRes.json();
|
||||
const world = await worldRes.json();
|
||||
|
||||
const container = document.getElementById('wanGeoMap');
|
||||
const w = container.clientWidth, h = container.clientHeight;
|
||||
const svg = d3.select('#wanGeoMap').append('svg').attr('width', w).attr('height', h);
|
||||
const proj = d3.geoNaturalEarth1().scale(w / 6.3).translate([w / 2, h / 2]);
|
||||
const path = d3.geoPath(proj);
|
||||
const countries = topojson.feature(world, world.objects.countries);
|
||||
|
||||
const countMap = Object.fromEntries((geo.countries || []).map(c => [c.country_code, c.total]));
|
||||
const maxCount = Math.max(...Object.values(countMap), 1);
|
||||
const colour = d3.scaleSequential(d3.interpolateReds).domain([0, maxCount]);
|
||||
|
||||
// Build numeric→alpha2 lookup using the same table from the DNS map
|
||||
const alpha2Num = {"AF":"004","AX":"008","AL":"008","DZ":"012","AS":"016","AD":"020","AO":"024","AI":"660","AQ":"010","AG":"028","AR":"032","AM":"051","AW":"533","AU":"036","AT":"040","AZ":"031","BS":"044","BH":"048","BD":"050","BB":"052","BY":"112","BE":"056","BZ":"084","BJ":"204","BM":"060","BT":"064","BO":"068","BA":"070","BW":"072","BV":"074","BR":"076","IO":"086","BN":"096","BG":"100","BF":"854","BI":"108","CV":"132","KH":"116","CM":"120","CA":"124","KY":"136","CF":"140","TD":"148","CL":"152","CN":"156","CX":"162","CC":"166","CO":"170","KM":"174","CG":"178","CD":"180","CK":"184","CR":"188","CI":"384","HR":"191","CU":"192","CW":"531","CY":"196","CZ":"203","DK":"208","DJ":"262","DM":"212","DO":"214","EC":"218","EG":"818","SV":"222","GQ":"226","ER":"232","EE":"233","SZ":"748","ET":"231","FK":"238","FO":"234","FJ":"242","FI":"246","FR":"250","GF":"254","PF":"258","TF":"260","GA":"266","GM":"270","GE":"268","DE":"276","GH":"288","GI":"292","GR":"300","GL":"304","GD":"308","GP":"312","GU":"316","GT":"320","GG":"831","GN":"324","GW":"624","GY":"328","HT":"332","HM":"334","VA":"336","HN":"340","HK":"344","HU":"348","IS":"352","IN":"356","ID":"360","IR":"364","IQ":"368","IE":"372","IM":"833","IL":"376","IT":"380","JM":"388","JP":"392","JE":"832","JO":"400","KZ":"398","KE":"404","KI":"296","KP":"408","KR":"410","KW":"414","KG":"417","LA":"418","LV":"428","LB":"422","LS":"426","LR":"430","LY":"434","LI":"438","LT":"440","LU":"442","MO":"446","MG":"450","MW":"454","MY":"458","MV":"462","ML":"466","MT":"470","MH":"584","MQ":"474","MR":"478","MU":"480","YT":"175","MX":"484","FM":"583","MD":"498","MC":"492","MN":"496","ME":"499","MS":"500","MA":"504","MZ":"508","MM":"104","NA":"516","NR":"520","NP":"524","NL":"528","NC":"540","NZ":"554","NI":"558","NE":"562","NG":"566","NU":"570","NF":"574","MK":"807","MP":"580","NO":"578","OM":"512","PK":"586","PW":"585","PS":"275","PA":"591","PG":"598","PY":"600","PE":"604","PH":"608","PN":"612","PL":"616","PT":"620","PR":"630","QA":"634","RE":"638","RO":"642","RU":"643","RW":"646","BL":"652","SH":"654","KN":"659","LC":"662","MF":"663","PM":"666","VC":"670","WS":"882","SM":"674","ST":"678","SA":"682","SN":"686","RS":"688","SC":"690","SL":"694","SG":"702","SX":"534","SK":"703","SI":"705","SB":"090","SO":"706","ZA":"710","GS":"239","SS":"728","ES":"724","LK":"144","SD":"729","SR":"740","SJ":"744","SE":"752","CH":"756","SY":"760","TW":"158","TJ":"762","TZ":"834","TH":"764","TL":"626","TG":"768","TK":"772","TO":"776","TT":"780","TN":"788","TR":"792","TM":"795","TC":"796","TV":"798","UG":"800","UA":"804","AE":"784","GB":"826","US":"840","UM":"581","UY":"858","UZ":"860","VU":"548","VE":"862","VN":"704","VG":"092","VI":"850","WF":"876","EH":"732","YE":"887","ZM":"894","ZW":"716"};
|
||||
const numAlpha2 = Object.fromEntries(Object.entries(alpha2Num).map(([a2, num]) => [num, a2]));
|
||||
|
||||
svg.append('g').selectAll('path')
|
||||
.data(countries.features)
|
||||
.join('path')
|
||||
.attr('d', path)
|
||||
.attr('fill', f => {
|
||||
const num = String(f.id).padStart(3, '0');
|
||||
const a2 = numAlpha2[num];
|
||||
const cnt = a2 ? countMap[a2] : 0;
|
||||
return cnt ? colour(cnt) : '#e5e7eb';
|
||||
})
|
||||
.attr('stroke', '#d1d5db').attr('stroke-width', 0.4)
|
||||
.append('title').text(f => {
|
||||
const num = String(f.id).padStart(3, '0');
|
||||
const a2 = numAlpha2[num];
|
||||
const cnt = a2 ? countMap[a2] : 0;
|
||||
return cnt ? `${a2}: ${cnt} attempts` : '';
|
||||
});
|
||||
|
||||
// Bubble circles for source IP locations
|
||||
const maxBubble = Math.max(...(geo.bubbles || []).map(b => b.total), 1);
|
||||
const rScale = d3.scaleSqrt().domain([1, maxBubble]).range([3, 24]);
|
||||
|
||||
svg.append('g').selectAll('circle')
|
||||
.data(geo.bubbles || [])
|
||||
.join('circle')
|
||||
.attr('cx', b => proj([b.lon, b.lat])[0])
|
||||
.attr('cy', b => proj([b.lon, b.lat])[1])
|
||||
.attr('r', b => rScale(b.total))
|
||||
.attr('fill', '#f43f5e')
|
||||
.attr('fill-opacity', 0.55)
|
||||
.attr('stroke', '#e11d48')
|
||||
.attr('stroke-width', 0.8)
|
||||
.append('title')
|
||||
.text(b => `${b.label}: ${b.total} attempts`);
|
||||
|
||||
// Legend
|
||||
const legend = document.getElementById('wan-geo-legend');
|
||||
(geo.countries || []).slice(0, 5).forEach(c => {
|
||||
const span = document.createElement('span');
|
||||
span.className = 'flex items-center gap-1';
|
||||
span.innerHTML = `<span class="w-2 h-2 rounded-full bg-rose-400 inline-block"></span>${c.country || c.country_code} (${c.total})`;
|
||||
legend.appendChild(span);
|
||||
});
|
||||
}
|
||||
|
||||
loadWanGeoMap();
|
||||
|
||||
function setWanFilter(field, value) {
|
||||
if (field === 'src_ip') document.getElementById('wan-filter-src').value = value;
|
||||
if (field === 'dst_port') document.getElementById('wan-filter-port').value = value;
|
||||
|
||||
@@ -11,6 +11,7 @@ urlpatterns = [
|
||||
path('api/chart/', views.ChartDataView.as_view(), name='routermon-chart'),
|
||||
path('api/geo/', views.GeoStatsView.as_view(), name='routermon-geo'),
|
||||
path('api/wan/chart/', views.WanChartDataView.as_view(), name='routermon-wan-chart'),
|
||||
path('api/wan/geo/', views.WanGeoStatsView.as_view(), name='routermon-wan-geo'),
|
||||
# Actions
|
||||
path('toggle/', views.ToggleView.as_view(), name='routermon-toggle'),
|
||||
path('status/', views.StatusStreamView.as_view(), name='routermon-status'),
|
||||
|
||||
@@ -367,6 +367,50 @@ class WanLivePartialView(View):
|
||||
return render(request, 'routermon/_live_wan.html', {'wan_events': qs[:60]})
|
||||
|
||||
|
||||
class WanGeoStatsView(View):
|
||||
"""Country-level aggregates + bubble data for WAN incoming events."""
|
||||
|
||||
def get(self, request):
|
||||
range_key, cfg, since = _get_range(request)
|
||||
qs = WanEvent.objects.filter(timestamp__gte=since)
|
||||
|
||||
countries = list(
|
||||
qs.exclude(country_code='')
|
||||
.values('country_code', 'country')
|
||||
.annotate(total=Count('id'))
|
||||
.order_by('-total')
|
||||
)
|
||||
|
||||
ip_agg = list(
|
||||
qs.values('src_ip')
|
||||
.annotate(total=Count('id'))
|
||||
.order_by('-total')[:300]
|
||||
)
|
||||
ip_set = [r['src_ip'] for r in ip_agg]
|
||||
|
||||
try:
|
||||
from nginxmon.models import IPGeoCache
|
||||
geo_cache = {
|
||||
c.ip: (c.lat, c.lon, c.country, c.city)
|
||||
for c in IPGeoCache.objects.filter(ip__in=ip_set, lat__isnull=False, is_private=False)
|
||||
}
|
||||
except Exception:
|
||||
geo_cache = {}
|
||||
|
||||
bubbles = []
|
||||
for row in ip_agg:
|
||||
c = geo_cache.get(row['src_ip'])
|
||||
if c and c[0] is not None and c[1] is not None:
|
||||
bubbles.append({
|
||||
'lat': round(float(c[0]), 2),
|
||||
'lon': round(float(c[1]), 2),
|
||||
'total': row['total'],
|
||||
'label': c[3] or c[2] or row['src_ip'],
|
||||
})
|
||||
|
||||
return JsonResponse({'countries': countries, 'bubbles': bubbles, 'range': range_key})
|
||||
|
||||
|
||||
class WanChartDataView(View):
|
||||
"""JSON — WAN event counts over time (for chart)."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user