From 62ac98ba214ede8193f5ad8d6aa1eedaa6e52f36 Mon Sep 17 00:00:00 2001 From: pavel Date: Thu, 30 Oct 2025 07:10:27 +0200 Subject: [PATCH] feat: startign with netbox 07:10 --- files/nats_registration_listener.py | 250 +++++++++++----------------- 1 file changed, 93 insertions(+), 157 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index b0d5494..1b02236 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,32 +1,13 @@ #!/usr/bin/env python3 """ -NATS Registration Listener (with NetBox hostname lookup for fox100, no extra deps) ----------------------------------------------------------------------------------- - -Behavior: - - Prints exactly one line per message: - [TS] product= mac= fw= [host= for fox100] - - Listener-only (no publish/reply). - - Payload-byte strict dedupe within a TTL window. - - Graceful shutdown + reconnect logging. - - For product=fox100: best-effort NetBox lookup by MAC -> interface -> device.name. - (Inline, sequential "as-is" call; no concurrency / worker pool.) - -NetBox (hardcoded as requested): - - Edit NB_TOKEN below. - - Uses stdlib urllib (no 'requests' / 'aiohttp'). - -CLI (same spirit as before; your entrypoint passes these): - --servers nats://host:4222 [one or more] - --subject v1.registration - --queue (optional) - --name registration-listener - --tls-ca /path/to/ca.pem - --tls-cert/--tls-key/--insecure - --include-subject - -Install deps in image: - pip install --no-cache-dir --upgrade nats-py +NATS Registration Listener (with NetBox hostname lookup for fox100 + timing + problem counter) +---------------------------------------------------------------------------------------------- +Adds: + - nb_problems= prefix on every stdout line + - Measures per-fox100 NetBox lookup duration (netbox_ms) + - Measures total latency from message arrival till final output (total_ms) + - Increments problem counter for each NetBox anomaly or lookup issue + - Adds iface_id=... to stderr diagnostics """ import argparse @@ -37,6 +18,7 @@ import signal import ssl import sys import hashlib +import time from time import monotonic from datetime import datetime, timezone from typing import Optional, Dict, Any @@ -51,43 +33,32 @@ import nats # NetBox hardcoded config # ========================= NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL -NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # <-- paste the real token here -NB_TIMEOUT = 3.0 # seconds per HTTP GET +NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided +NB_TIMEOUT = 3.0 # seconds per HTTP GET + +# Global counter for any NetBox-related errors/anomalies +NB_PROBLEM_COUNTER = 0 +NB_PROBLEM_LOCK = asyncio.Lock() # ========================= -# Arg parsing (mirrors original) +# Arg parsing # ========================= def parse_args(): p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.") - p.add_argument( - "--servers", - nargs="+", - help="One or more NATS server URLs (e.g., nats://127.0.0.1:4222).", - default=["nats://127.0.0.1:4222"], - ) - p.add_argument( - "--subject", - help="NATS subject to subscribe to.", - default="registration", - ) - p.add_argument("--queue", help="Optional queue group name.", default=None) - p.add_argument("--name", help="Client connection name.", default="registration-listener") - - # Auth - p.add_argument("--creds", help="Path to .creds file (JWT + NKey).") - p.add_argument("--user", help="Username.") - p.add_argument("--password", help="Password.") - p.add_argument("--token", help="Auth token.") - - # TLS - p.add_argument("--tls-ca", help="Path to CA certificate for TLS.") - p.add_argument("--tls-cert", help="Path to client certificate for TLS.") - p.add_argument("--tls-key", help="Path to client key for TLS.") - p.add_argument("--insecure", action="store_true", help="Disable TLS hostname verification.") - - # Misc - p.add_argument("--include-subject", action="store_true", help="Append subject=... to each output line.") + p.add_argument("--servers", nargs="+", default=["nats://127.0.0.1:4222"]) + p.add_argument("--subject", default="registration") + p.add_argument("--queue", default=None) + p.add_argument("--name", default="registration-listener") + p.add_argument("--creds") + p.add_argument("--user") + p.add_argument("--password") + p.add_argument("--token") + p.add_argument("--tls-ca") + p.add_argument("--tls-cert") + p.add_argument("--tls-key") + p.add_argument("--insecure", action="store_true") + p.add_argument("--include-subject", action="store_true") return p.parse_args() @@ -97,23 +68,18 @@ def parse_args(): def make_ssl_context(args) -> Optional[ssl.SSLContext]: if not any([args.tls_ca, args.tls_cert, args.tls_key]) and not any(url.startswith("tls://") for url in args.servers): return None - ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) if args.tls_ca: ctx.load_verify_locations(args.tls_ca) - if args.tls_cert and args.tls_key: ctx.load_cert_chain(certfile=args.tls_cert, keyfile=args.tls_key) - if args.insecure: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE - return ctx def ts() -> str: - # local time with offset like your previous output return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z") @@ -127,50 +93,23 @@ def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]: def extract_fields(obj: Dict[str, Any]): - """ - Expected structure (from examples): - { "data": { - "productName": "...", - "ethernetInterfaces": { "eth0": { "macAddress": "..." } }, - "firmwareVersion": { "active": "..." } - }, - "metadata": { "type": "REGISTRATION_REQUEST" } - } - """ root = obj - if isinstance(root.get("data"), dict): - d = root["data"] - else: - d = root - + d = root["data"] if isinstance(root.get("data"), dict) else root product = d.get("productName") or "-" - fw_active = "-" - try: - fw_active = d.get("firmwareVersion", {}).get("active") or "-" - except Exception: - pass - + fw_active = d.get("firmwareVersion", {}).get("active") or "-" mac = "-" - try: - eths = d.get("ethernetInterfaces", {}) - if isinstance(eths, dict): - chosen = pick_first_interface(eths) - if chosen and isinstance(chosen.get("macAddress"), str): - mac = chosen["macAddress"] - except Exception: - pass - + eths = d.get("ethernetInterfaces", {}) + if isinstance(eths, dict): + chosen = pick_first_interface(eths) + if chosen and isinstance(chosen.get("macAddress"), str): + mac = chosen["macAddress"] return product, mac, fw_active # ========================= -# NetBox lookup (stdlib urllib) +# NetBox lookup (urllib) # ========================= def normalize_mac(mac: str) -> Optional[str]: - """ - Return lowercase colon-separated MAC or None if invalid-ish. - Accepts colon, dash, or bare hex and tries to normalize. - """ if not mac or not isinstance(mac, str): return None s = mac.strip().lower().replace("-", ":") @@ -192,10 +131,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op if resp.status != 200: return None, resp.status data = resp.read() - try: - return json.loads(data.decode("utf-8", errors="replace")), 200 - except Exception: - return None, 200 + return json.loads(data.decode("utf-8", errors="replace")), 200 except HTTPError as e: return None, getattr(e, "code", 599) except URLError: @@ -204,18 +140,21 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op return None, 597 -def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]: +async def nb_problem(log_status, msg: str): + """Increment counter and log a problem line.""" + global NB_PROBLEM_COUNTER + async with NB_PROBLEM_LOCK: + NB_PROBLEM_COUNTER += 1 + await log_status(msg) + + +def nb_get_hostname_by_mac(mac: str, log_status) -> (Optional[str], Optional[int]): """ - Steps: - 1) GET /api/dcim/mac-addresses/?mac_address= - 2) If assigned to dcim.interface -> GET /api/dcim/interfaces/{id}/ - 3) Return device.name - Emits concise diagnostics to stderr via log_status on edge cases. + Return (hostname, iface_id or None) """ mac_norm = normalize_mac(mac) if not mac_norm: - # invalid or missing mac; keep stdout clean - return None + return None, None base = NB_URL.rstrip("/") h = { @@ -224,42 +163,43 @@ def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]: "Authorization": f"Token {NB_TOKEN}", } - # 1) find MAC record(s) + # Step 1: MAC lookup data, code = http_get_json(f"{base}/api/dcim/mac-addresses/", params={"mac_address": mac_norm, "limit": "2"}, headers=h) if code != 200 or not data: - # network/HTTP issue or parse fail - asyncio.create_task(log_status(f"[{ts()}] nb: mac query http={code} mac={mac_norm}")) - return None + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac query http={code} mac={mac_norm}")) + return None, None results = (data or {}).get("results") or [] if not results: - asyncio.create_task(log_status(f"[{ts()}] nb: mac not found mac={mac_norm}")) - return None + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac not found mac={mac_norm}")) + return None, None if len(results) > 1: - asyncio.create_task(log_status(f"[{ts()}] nb: multiple mac records mac={mac_norm}")) + # no iface_id yet; fetch it below + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: multiple mac records mac={mac_norm}")) rec = results[0] aot = (rec.get("assigned_object_type") or "").strip() aoid = rec.get("assigned_object_id") if not aot or aoid is None: - asyncio.create_task(log_status(f"[{ts()}] nb: mac unassigned mac={mac_norm}")) - return None + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac unassigned mac={mac_norm}")) + return None, None if aot != "dcim.interface": - asyncio.create_task(log_status(f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm}")) - return None + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm} iface_id={aoid}")) + return None, aoid - # 2) fetch interface -> device + # Step 2: Interface → Device iface, code2 = http_get_json(f"{base}/api/dcim/interfaces/{aoid}/", headers=h) if code2 != 200 or not iface: - asyncio.create_task(log_status(f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}")) - return None + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}")) + return None, aoid dev = iface.get("device") or {} host = dev.get("name") or dev.get("display") if not host: - asyncio.create_task(log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}")) - return None - return host + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface has no device iface_id={aoid}")) + return None, aoid + + return host, aoid # ========================= @@ -275,13 +215,9 @@ async def main(): async with print_lock: print(s, file=sys.stderr, flush=True) - # --- Strict read-only dedupe (payload bytes) --- - DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) # you can tune via env if you want; otherwise 2.0s + DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) recent_payloads: Dict[bytes, float] = {} - reconnect_time_wait = 2 - max_reconnect_attempts = -1 - async def disconnected_cb(): await log_status(f"[{ts()}] Disconnected from NATS.") @@ -298,8 +234,8 @@ async def main(): servers=args.servers, name=args.name, allow_reconnect=True, - reconnect_time_wait=reconnect_time_wait, - max_reconnect_attempts=max_reconnect_attempts, + reconnect_time_wait=2, + max_reconnect_attempts=-1, disconnected_cb=disconnected_cb, reconnected_cb=reconnected_cb, error_cb=error_cb, @@ -312,10 +248,9 @@ async def main(): ) async def message_handler(msg: nats.aio.msg.Msg): - now_s = ts() + t_start = monotonic() payload = msg.data - # --- dedupe by exact payload bytes --- digest = hashlib.blake2b(payload, digest_size=16).digest() nowm = monotonic() exp = recent_payloads.get(digest) @@ -325,10 +260,9 @@ async def main(): if len(recent_payloads) > 4096: cutoff = nowm for k in list(recent_payloads.keys()): - if recent_payloads.get(k, 0) <= cutoff: + if recent_payloads[k] <= cutoff: recent_payloads.pop(k, None) - # parse product = mac = fw = "-" try: text = payload.decode("utf-8", errors="replace") @@ -337,39 +271,43 @@ async def main(): except Exception: pass - # optional NetBox lookup for fox100 (inline, sequential) host_suffix = "" + netbox_time_ms = 0.0 if product == "fox100": + nb_start = time.perf_counter() try: - host = nb_get_hostname_by_mac(mac=mac, log_status=log_status) + host, iface_id = nb_get_hostname_by_mac(mac=mac, log_status=log_status) if host: host_suffix = f" host={host}" - else: - # minimal diag already sent to stderr by helper - pass + if iface_id and not host: + host_suffix += f" iface_id={iface_id}" except Exception as e: - await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") + await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") + netbox_time_ms = (time.perf_counter() - nb_start) * 1000 - # print one compact line - line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}" + total_ms = (monotonic() - t_start) * 1000 + + async with NB_PROBLEM_LOCK: + nb_problems_snapshot = NB_PROBLEM_COUNTER + + line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}" + if product == "fox100": + line += f" netbox_ms={netbox_time_ms:.1f} total_ms={total_ms:.1f}" if args.include_subject: line += f" subject={msg.subject}" - async with print_lock: - sys.stdout.write(line + "\n") - sys.stdout.flush() - # subscribe + async with print_lock: + print(line, flush=True) + if args.queue: await nc.subscribe(args.subject, queue=args.queue, cb=message_handler) else: await nc.subscribe(args.subject, cb=message_handler) await log_status( - f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} " - f"| mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled nb=on" + f"[{ts()}] Listening on subject '{args.subject}' (queue={args.queue or '-'}) via {args.servers} | nb=on" ) - # graceful shutdown stop_event = asyncio.Event() def handle_signal(*_): @@ -384,10 +322,8 @@ async def main(): signal.signal(s, lambda *_: handle_signal()) await stop_event.wait() - try: - await nc.drain() - finally: - await nc.close() + await nc.drain() + await nc.close() if __name__ == "__main__":