diff --git a/routermon/receiver.py b/routermon/receiver.py index 7c7c051..f20cb65 100644 --- a/routermon/receiver.py +++ b/routermon/receiver.py @@ -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: diff --git a/routermon/templates/routermon/dashboard.html b/routermon/templates/routermon/dashboard.html index e46a0f0..7c7ccbb 100644 --- a/routermon/templates/routermon/dashboard.html +++ b/routermon/templates/routermon/dashboard.html @@ -292,6 +292,15 @@ + +
+
+
Source Geography — {{ range_label }}
+
+
+
+
+
{ 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 = `${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; diff --git a/routermon/urls.py b/routermon/urls.py index 28858e6..bde0429 100644 --- a/routermon/urls.py +++ b/routermon/urls.py @@ -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'), diff --git a/routermon/views.py b/routermon/views.py index a72e086..9daa14f 100644 --- a/routermon/views.py +++ b/routermon/views.py @@ -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)."""