mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
junvandCopilot
60dd5aeffb
Stop sending further API requests mid-batch once the backoff is triggered, saving remaining missing IPs as empty placeholders immediately instead of waiting for them to timeout one-by-one. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
472 lines
18 KiB
Python
472 lines
18 KiB
Python
"""
|
|
UDP syslog receiver for routermon.
|
|
|
|
Architecture:
|
|
- One background thread owns the UDP socket and enqueues parsed events.
|
|
- A separate writer thread drains the queue and bulk-saves to the DB every
|
|
FLUSH_INTERVAL seconds (or when the queue reaches FLUSH_SIZE).
|
|
- A single geo-enrichment worker drains a bounded queue so WAN scan volumes
|
|
don't create unbounded thread churn.
|
|
- Only one Gunicorn worker binds the socket (EADDRINUSE is silently ignored
|
|
in the others — matches the project's existing in-process APScheduler style).
|
|
|
|
dnsmasq reply correlation:
|
|
- A small in-memory dict `_pending` maps `domain` → (pk, timestamp) for
|
|
outstanding query rows awaiting a reply. Entries expire after REPLY_TTL
|
|
seconds so stale pending records don't accumulate.
|
|
"""
|
|
import errno
|
|
import logging
|
|
import queue
|
|
import socket
|
|
import threading
|
|
import time
|
|
from datetime import timedelta
|
|
|
|
from django.utils import timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FLUSH_INTERVAL = 2 # seconds between writer flushes
|
|
FLUSH_SIZE = 100 # flush immediately when queue reaches this size
|
|
REPLY_TTL = 10 # seconds to keep a query in _pending for reply matching
|
|
MAX_QUEUE = 10_000 # drop oldest items if queue exceeds this (backpressure)
|
|
MAX_GEO_QUEUE = 500 # bounded geo-enrichment queue (drop excess rather than thread-spawn)
|
|
|
|
_receiver_thread: threading.Thread | None = None
|
|
_writer_thread: threading.Thread | None = None
|
|
_geo_thread: threading.Thread | None = None
|
|
_sock: socket.socket | None = None
|
|
_queue: queue.Queue = queue.Queue(maxsize=MAX_QUEUE)
|
|
_geo_queue: queue.Queue = queue.Queue(maxsize=MAX_GEO_QUEUE)
|
|
|
|
# in-memory pending dict: domain → (query_pk, created_at)
|
|
_pending: dict = {}
|
|
_pending_lock = threading.Lock()
|
|
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
def start_receiver(port: int):
|
|
"""
|
|
Attempt to start the UDP receiver on `port`. Safe to call from every
|
|
Gunicorn worker — only the first to bind succeeds; others silently skip.
|
|
"""
|
|
global _receiver_thread, _writer_thread, _geo_thread, _sock
|
|
|
|
if _receiver_thread and _receiver_thread.is_alive():
|
|
return # already running in this process
|
|
|
|
try:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
# No SO_REUSEADDR: we intentionally want only one worker to bind.
|
|
# Other workers receive EADDRINUSE and skip cleanly.
|
|
sock.bind(('0.0.0.0', port))
|
|
sock.settimeout(1.0)
|
|
_sock = sock
|
|
print(f'routermon: UDP syslog receiver bound on port {port}', flush=True)
|
|
logger.info('routermon: UDP syslog receiver bound on port %d', port)
|
|
except OSError as exc:
|
|
if exc.errno == errno.EADDRINUSE:
|
|
logger.debug('routermon: port %d already bound (another worker), skipping', port)
|
|
else:
|
|
import sys
|
|
print(f'routermon: failed to bind UDP port {port}: {exc}', file=sys.stderr, flush=True)
|
|
logger.error('routermon: failed to bind UDP port %d: %s', port, exc)
|
|
return
|
|
|
|
_geo_thread = threading.Thread(target=_geo_worker_loop, daemon=True, name='routermon-geo')
|
|
_geo_thread.start()
|
|
|
|
_writer_thread = threading.Thread(target=_writer_loop, daemon=True, name='routermon-writer')
|
|
_writer_thread.start()
|
|
|
|
_receiver_thread = threading.Thread(
|
|
target=_receiver_loop, args=(sock,), daemon=True, name='routermon-receiver'
|
|
)
|
|
_receiver_thread.start()
|
|
|
|
|
|
def stop_receiver():
|
|
global _sock
|
|
if _sock:
|
|
try:
|
|
_sock.close()
|
|
except Exception:
|
|
pass
|
|
_sock = None
|
|
|
|
|
|
# ── Socket receiver loop ──────────────────────────────────────────────────────
|
|
|
|
def _receiver_loop(sock: socket.socket):
|
|
from .parser import parse_line
|
|
|
|
logger.info('routermon: receiver loop started')
|
|
while True:
|
|
try:
|
|
data, _ = sock.recvfrom(4096)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
# Socket closed (shutdown)
|
|
break
|
|
try:
|
|
line = data.decode('utf-8', errors='replace').rstrip('\n\r')
|
|
result = parse_line(line)
|
|
if result is not None:
|
|
_enqueue(result)
|
|
except Exception as exc:
|
|
logger.debug('routermon: parse error: %s', exc)
|
|
|
|
logger.info('routermon: receiver loop stopped')
|
|
|
|
|
|
def _enqueue(result):
|
|
try:
|
|
_queue.put_nowait(result)
|
|
except queue.Full:
|
|
# Drop oldest item to make room (keep most recent)
|
|
try:
|
|
_queue.get_nowait()
|
|
_queue.put_nowait(result)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── Writer loop ───────────────────────────────────────────────────────────────
|
|
|
|
def _writer_loop():
|
|
logger.info('routermon: writer loop started')
|
|
batch = []
|
|
last_flush = time.monotonic()
|
|
|
|
while True:
|
|
# Collect items until flush threshold or timeout
|
|
deadline = last_flush + FLUSH_INTERVAL
|
|
while time.monotonic() < deadline and len(batch) < FLUSH_SIZE:
|
|
try:
|
|
item = _queue.get(timeout=max(0.1, deadline - time.monotonic()))
|
|
batch.append(item)
|
|
except queue.Empty:
|
|
break
|
|
|
|
if batch:
|
|
try:
|
|
_flush(batch)
|
|
except Exception as exc:
|
|
logger.error('routermon: flush error: %s', exc, exc_info=True)
|
|
batch = []
|
|
|
|
last_flush = time.monotonic()
|
|
|
|
logger.info('routermon: writer loop stopped')
|
|
|
|
|
|
def _flush(batch: list):
|
|
"""Persist a batch of ParseResult objects to the database."""
|
|
from .models import RouterMonSettings, DnsQuery, DhcpLease, WanEvent
|
|
|
|
settings = RouterMonSettings.get()
|
|
if not settings.enabled:
|
|
return
|
|
|
|
excluded = _build_excluded_set(settings.excluded_clients)
|
|
now = timezone.now()
|
|
|
|
# --- DHCP leases (upsert immediately) ---
|
|
dhcp_events = [r.dhcp for r in batch if r.dhcp]
|
|
for ev in dhcp_events:
|
|
DhcpLease.objects.update_or_create(
|
|
ip=ev.ip,
|
|
defaults={'mac': ev.mac, 'hostname': ev.hostname},
|
|
)
|
|
|
|
# --- DNS queries ---
|
|
hostname_cache = {
|
|
lease.ip: lease.hostname
|
|
for lease in DhcpLease.objects.filter(
|
|
ip__in={r.query.client_ip for r in batch if r.query}
|
|
)
|
|
}
|
|
|
|
query_objs = []
|
|
for r in batch:
|
|
if not r.query:
|
|
continue
|
|
q = r.query
|
|
if q.client_ip in excluded:
|
|
continue
|
|
obj = DnsQuery(
|
|
timestamp=now,
|
|
client_ip=q.client_ip,
|
|
client_name=hostname_cache.get(q.client_ip, ''),
|
|
domain=q.domain,
|
|
query_type=q.query_type,
|
|
)
|
|
query_objs.append(obj)
|
|
|
|
if query_objs:
|
|
created = DnsQuery.objects.bulk_create(query_objs)
|
|
# Register created queries in pending dict for reply correlation
|
|
with _pending_lock:
|
|
for obj in created:
|
|
_pending[obj.domain] = (obj.pk, now)
|
|
|
|
# --- DNS replies: correlate with pending queries ---
|
|
reply_events = [r.reply for r in batch if r.reply]
|
|
if reply_events:
|
|
_apply_replies(reply_events, now)
|
|
|
|
# Expire old pending entries
|
|
_expire_pending(now)
|
|
|
|
# --- WAN events ---
|
|
wan_events = [r.wan_event for r in batch if r.wan_event]
|
|
if wan_events:
|
|
wan_objs = [
|
|
WanEvent(
|
|
timestamp=now,
|
|
src_ip=ev.src_ip,
|
|
protocol=ev.protocol,
|
|
dst_port=ev.dst_port,
|
|
src_port=ev.src_port,
|
|
)
|
|
for ev in wan_events
|
|
]
|
|
WanEvent.objects.bulk_create(wan_objs)
|
|
# 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 and ev.src_ip not in _geo_skip:
|
|
seen_ips.add(ev.src_ip)
|
|
try:
|
|
_geo_queue.put_nowait(('wan', ev.src_ip))
|
|
except queue.Full:
|
|
pass # drop silently under high scan volume
|
|
|
|
# Touch last_received_at once per flush (coalesced)
|
|
RouterMonSettings.objects.filter(pk=1).update(last_received_at=now)
|
|
|
|
# Enqueue DNS geo enrichment (deduplicated by IP)
|
|
pks_needing_geo = list(
|
|
DnsQuery.objects.filter(
|
|
resolved_ip__isnull=False, country='', timestamp__gte=now - timedelta(minutes=5)
|
|
).values_list('pk', flat=True)[:200]
|
|
)
|
|
for pk in pks_needing_geo:
|
|
try:
|
|
_geo_queue.put_nowait(('dns', pk))
|
|
except queue.Full:
|
|
break
|
|
|
|
|
|
def _apply_replies(reply_events, now):
|
|
from .models import DnsQuery
|
|
|
|
with _pending_lock:
|
|
for ev in reply_events:
|
|
entry = _pending.pop(ev.domain, None)
|
|
if entry is None:
|
|
continue
|
|
pk, _ = entry
|
|
updates = {}
|
|
if ev.is_nxdomain:
|
|
updates['is_nxdomain'] = True
|
|
if ev.resolved_ip:
|
|
updates['resolved_ip'] = ev.resolved_ip
|
|
if updates:
|
|
DnsQuery.objects.filter(pk=pk).update(**updates)
|
|
|
|
|
|
def _expire_pending(now):
|
|
cutoff = now - timedelta(seconds=REPLY_TTL)
|
|
with _pending_lock:
|
|
expired = [domain for domain, (_, ts) in _pending.items() if ts < cutoff]
|
|
for domain in expired:
|
|
del _pending[domain]
|
|
|
|
|
|
# ── 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
|
|
|
|
# Circuit breaker: if ip-api.com returns empty for N+ consecutive large chunks,
|
|
# back off for GEO_BACKOFF_SECS before trying again.
|
|
_geo_consecutive_misses: int = 0
|
|
_geo_backoff_until: float = 0.0
|
|
_GEO_MISS_THRESHOLD = 2 # consecutive all-empty chunks before backoff
|
|
_GEO_BACKOFF_SECS = 300.0 # 5-minute backoff when unreachable
|
|
|
|
|
|
def _geo_worker_loop():
|
|
"""
|
|
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.
|
|
|
|
Circuit breaker: after _GEO_MISS_THRESHOLD consecutive all-empty API
|
|
responses, all remaining missing IPs are saved as empty placeholders and
|
|
API calls are suppressed for _GEO_BACKOFF_SECS (5 min) to avoid a
|
|
timeout-storm when ip-api.com is unreachable.
|
|
"""
|
|
global _geo_consecutive_misses, _geo_backoff_until
|
|
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:
|
|
first = _geo_queue.get(timeout=5)
|
|
except queue.Empty:
|
|
continue
|
|
|
|
# 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:
|
|
# 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:
|
|
# Circuit breaker: if ip-api.com has been unresponsive, skip API
|
|
# calls and immediately cache all missing IPs as empty so they
|
|
# are not retried until the backoff expires.
|
|
in_backoff = time.monotonic() < _geo_backoff_until
|
|
if in_backoff:
|
|
IPGeoCache.objects.bulk_create(
|
|
[IPGeoCache(ip=ip) for ip in missing],
|
|
ignore_conflicts=True,
|
|
)
|
|
_geo_skip.update(missing)
|
|
if len(_geo_skip) > _GEO_SKIP_MAX:
|
|
_geo_skip.clear()
|
|
else:
|
|
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()
|
|
|
|
# Circuit-breaker tracking: a large chunk with zero results
|
|
# almost certainly means the API is unreachable.
|
|
if len(chunk) >= 5 and not results:
|
|
_geo_consecutive_misses += 1
|
|
if _geo_consecutive_misses >= _GEO_MISS_THRESHOLD:
|
|
_geo_backoff_until = time.monotonic() + _GEO_BACKOFF_SECS
|
|
logger.warning(
|
|
'routermon: ip-api.com unreachable (%d consecutive misses) — '
|
|
'backing off for %.0fs',
|
|
_geo_consecutive_misses, _GEO_BACKOFF_SECS,
|
|
)
|
|
_geo_consecutive_misses = 0
|
|
# Save remaining uncached IPs and stop the loop early.
|
|
remaining = missing[i + 100:]
|
|
if remaining:
|
|
IPGeoCache.objects.bulk_create(
|
|
[IPGeoCache(ip=ip) for ip in remaining
|
|
if ip not in geo_cache],
|
|
ignore_conflicts=True,
|
|
)
|
|
_geo_skip.update(remaining)
|
|
break
|
|
elif results:
|
|
_geo_consecutive_misses = 0
|
|
|
|
# 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.warning('routermon: geo batch error: %s', exc)
|
|
|
|
|
|
def _build_excluded_set(text: str) -> set:
|
|
result = set()
|
|
for line in (text or '').splitlines():
|
|
entry = line.strip()
|
|
if entry and not entry.startswith('#'):
|
|
result.add(entry)
|
|
return result
|
|
|