diff --git a/files/nats_registration_listener-0323.py b/files/nats_registration_listener-0323.py new file mode 100644 index 0000000..4db4ffe --- /dev/null +++ b/files/nats_registration_listener-0323.py @@ -0,0 +1,997 @@ +#!/usr/bin/env python3 +""" +NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing + problem counter) +------------------------------------------------------------------------------------------- +- One device GET (custom_fields.action_next), no duplicate fetch +- Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting +- For fox100: + * If action_next is empty/absent -> print a simple stdout note and do nothing else + * If action_next present and action_last != action_next -> publish task_name=action_next + and on success set action_last=action_next and action_next_timestamp=now_epoch + * If action_next present and action_last == action_next -> publish only if + (now_epoch - action_next_timestamp) >= 600; if timestamp missing/invalid -> allow publish + and on success set action_last=action_next and action_next_timestamp=now_epoch + * If action_next present -> prepend 3x ASCII BEL to stdout line (kept behavior) +""" + +import argparse +import asyncio +import json +import os +import signal +import ssl +import sys +import hashlib +import time +from time import monotonic +from datetime import datetime, timezone, timedelta +from typing import Optional, Dict, Any, Tuple, Set +from urllib.parse import urlencode +from urllib.request import Request, urlopen +from urllib.error import URLError, HTTPError +import base64 # for RabbitMQ Basic Auth +from pathlib import Path + +import nats + + +# ========================= +# NetBox hardcoded config +# ========================= +NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL +NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided +NB_TIMEOUT = 6.0 # seconds per HTTP GET + +# Cache for MAC -> NetBox lookup result (seconds). Keeps NetBox load down under chatty devices. +NB_LOOKUP_CACHE_TTL = float(os.environ.get("NB_LOOKUP_CACHE_TTL", "60.0")) +NB_LOOKUP_CACHE = {} # mac_norm -> (expires_monotonic, cached_tuple) + +# ========================= +# RabbitMQ hardcoded config +# ========================= +RMQ_HOST = "10.210.12.2" +RMQ_PORT = 15672 +RMQ_USER = "admin" +RMQ_PASS = "change_me" +RMQ_VHOST = "app" +RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) +RMQ_EXCHANGE_DELAYED = "deviceconfig.delayed" # delayed exchange (x-delayed-message) +RMQ_ROUTING_KEY = "deviceconfig" +RMQ_TIMEOUT = 5.0 + +# ---- Human-editable delay (milliseconds). Set to 0 to disable delay. +# Example: 600000 = 10 minutes +RMQ_DELAY_MS = 15000 + +# Posture analyzer gate: skip re-running sot-updater-scheduler if recently run (seconds) +posture_analyzer = "sot-updater-scheduler" +sot_timeout = 300 + +# Product -> Tag slug mapping (kept unchanged, though not used now) +PRODUCT_TAG_SLUG = { + "fox100": "fox100-auto-upgrade-latest", + # "fox200": "fox200-auto-upgrade-latest", +} + + + + +# MACs to ignore completely when seen in registrations +IGNORE_MACS_RAW = [ + "AA:BB:CC:DD:EE:FF", + "D0:6C:37:01:26:02", + "D0:6C:37:01:25:B2", + "D0:6C:37:01:25:EA", + "D0:6C:37:01:25:F2", + "D0:6C:37:01:26:02", + "D0:6C:37:01:26:1E", + "D0:6C:37:01:26:3A", + "D0:6C:37:01:26:D2", + "D0:6C:37:01:26:D6", + "D0:6C:37:01:26:E2", + "D0:6C:37:01:0C:C0", + "D0:6C:37:00:91:88", + "C4:93:00:4E:96:0C", + "D0:6C:37:01:0C:78", + "C4:93:00:51:9A:12", + "D0:6C:37:00:BD:B0", + "C4:93:00:4E:97:B0", + #"D0:6C:37:00:A9:68", #### 97227 + + +] + +# Global counter for any NetBox-related problems +NB_PROBLEM_COUNTER = 0 +NB_PROBLEM_LOCK = asyncio.Lock() + +# Structured event log output (JSON Lines) +EVENT_LOG_PATH = os.environ.get("EVENT_LOG_PATH", "/opt/containers/nats-registration-listener/logs/registration_events.jsonl") + + +# ========================= +# 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="+", 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() + + +# ========================= +# Helpers +# ========================= +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: + return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z") + + +def ts_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def append_jsonl(path: str, obj: Dict[str, Any]): + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + with p.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + "\n") + + +def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if "eth0" in eths and isinstance(eths["eth0"], dict): + return eths["eth0"] + for name in sorted(eths.keys()): + if isinstance(eths[name], dict): + return eths[name] + return None + + +def extract_fields(obj: Dict[str, Any]): + root = obj + d = root["data"] if isinstance(root.get("data"), dict) else root + product = d.get("productName") or "-" + fw_active = d.get("firmwareVersion", {}).get("active") or "-" + mac = "-" + 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 + + +def _parse_possible_event_epoch(value: Any) -> Optional[int]: + try: + if value is None: + return None + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + v = float(value) + if v > 1e12: + v = v / 1000.0 + if v > 0: + return int(v) + return None + if isinstance(value, str): + s = value.strip() + if not s: + return None + if s.isdigit(): + v = float(s) + if v > 1e12: + v = v / 1000.0 + if v > 0: + return int(v) + return None + s2 = s.replace("Z", "+00:00") + try: + return int(datetime.fromisoformat(s2).timestamp()) + except Exception: + pass + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S%z", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"): + try: + dt = datetime.strptime(s, fmt) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return int(dt.timestamp()) + except Exception: + pass + except Exception: + return None + return None + + +def extract_registration_age_s(obj: Dict[str, Any]) -> Optional[int]: + candidate_keys = { + "timestamp", "ts", "time", "event_time", "eventtime", + "event_ts", "eventtimestamp", "created_at", "createdat", + "published_at", "publishedat", "sent_at", "sentat", + "received_at", "receivedat", + } + + def walk(node: Any) -> Optional[int]: + if isinstance(node, dict): + for k, v in node.items(): + ks = str(k).strip().lower().replace("-", "_") + if ks in candidate_keys: + parsed = _parse_possible_event_epoch(v) + if parsed is not None: + return parsed + for v in node.values(): + parsed = walk(v) + if parsed is not None: + return parsed + elif isinstance(node, list): + for item in node: + parsed = walk(item) + if parsed is not None: + return parsed + return None + + epoch = walk(obj) + if epoch is None: + return None + age_s = int(time.time()) - int(epoch) + if age_s < 0: + return 0 + return age_s + + +# ========================= +# NetBox lookup (urllib) +# ========================= +def normalize_mac(mac: str) -> Optional[str]: + if not mac or not isinstance(mac, str): + return None + s = mac.strip().lower().replace("-", ":") + hex_only = "".join(ch for ch in s if ch in "0123456789abcdef") + if len(hex_only) == 12: + return ":join".replace(":", "").join([":".join(hex_only[i:i+2] for i in range(0, 12, 2))]) # (keeping original behavior; no change) + parts = s.split(":") + if len(parts) == 6 and all(len(p) == 2 and all(c in "0123456789abcdef" for c in p) for p in parts): + return s + return None + + +IGNORE_MACS = {m for m in (normalize_mac(x) for x in IGNORE_MACS_RAW) if m} + + +def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, timeout: float = NB_TIMEOUT): + if params: + url = f"{url}?{urlencode(params)}" + req = Request(url, headers=headers or {}, method="GET") + try: + with urlopen(req, timeout=timeout) as resp: + if resp.status != 200: + return None, resp.status + data = resp.read() + return json.loads(data.decode("utf-8", errors="replace")), 200 + except HTTPError as e: + return None, getattr(e, "code", 599) + except URLError: + return None, 598 + except Exception: + return None, 597 + + +# RabbitMQ management API POST helper (basic auth; JSON in/out) +def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, timeout: float = RMQ_TIMEOUT): + body = json.dumps(payload_obj).encode("utf-8") + headers = {"Content-Type": "application/json"} + if user and password: + token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii") + headers["Authorization"] = f"Basic {token}" + req = Request(url, data=body, headers=headers, method="POST") + try: + with urlopen(req, timeout=timeout) as resp: + data = resp.read() + try: + return json.loads(data.decode("utf-8", errors="replace")), resp.status + except Exception: + return None, resp.status + except HTTPError as e: + return None, getattr(e, "code", 599) + except URLError: + return None, 598 + except Exception: + return None, 597 + + +# NetBox PATCH helper (JSON in/out) +def http_patch_json(url: str, payload_obj: Dict[str, Any], headers: Optional[Dict[str, str]] = None, timeout: float = NB_TIMEOUT): + body = json.dumps(payload_obj).encode("utf-8") + h = dict(headers or {}) + h["Content-Type"] = "application/json" + req = Request(url, data=body, headers=h, method="PATCH") + try: + with urlopen(req, timeout=timeout) as resp: + data = resp.read() + try: + return json.loads(data.decode("utf-8", errors="replace")), resp.status + except Exception: + return None, resp.status + except HTTPError as e: + return None, getattr(e, "code", 599) + except URLError: + return None, 598 + except Exception: + return None, 597 + + +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_lookup_device_by_mac(mac: str, log_status) -> Tuple[ + Optional[str], Optional[int], Optional[int], Optional[Any], Optional[Any], Optional[Any], Optional[Any], Optional[Any] +]: + """ + Resolve MAC -> (hostname, iface_id, device_id, action_next, action_last, action_next_timestamp, action_state, sot_ts) + - Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail). + - If device detail fetch fails, returns host/id with custom_fields as None. + """ + mac_norm = normalize_mac(mac) + if not mac_norm: + return None, None, None, None, None, None, None, None + + # Step 0: short TTL cache (avoid repeated NetBox GETs for chatty devices) + if NB_LOOKUP_CACHE_TTL > 0: + nowm = monotonic() + cached = NB_LOOKUP_CACHE.get(mac_norm) + if cached: + exp, val = cached + if exp > nowm: + host, iface_id, dev_id, action_next, action_last, action_next_timestamp, action_state, sot_ts = val + return host, iface_id, dev_id, action_next, action_last, action_next_timestamp, action_state, sot_ts + NB_LOOKUP_CACHE.pop(mac_norm, None) + + base = NB_URL.rstrip("/") + h = { + "Accept": "application/json", + "Content-Type": "application/json", + "Authorization": f"Token {NB_TOKEN}", + } + + # Step 1: MAC lookup + data, code = http_get_json( + f"{base}/api/dcim/mac-addresses/", + params={"mac_address": mac_norm, "limit": "2", "fields": "assigned_object_type,assigned_object_id"}, + headers=h, + ) + if code == 400: + 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: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac query http={code} mac={mac_norm}")) + return None, None, None, None, None, None, None, None + + results = (data or {}).get("results") or [] + if not results: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac not found mac={mac_norm}")) + return None, None, None, None, None, None, None, None + + rec = results[0] + aot = (rec.get("assigned_object_type") or "").strip() + aoid = rec.get("assigned_object_id") + + if len(results) > 1: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: multiple mac records mac={mac_norm} iface_id={aoid if aoid is not None else '-'}")) + + if not aot or aoid is None: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac unassigned mac={mac_norm}")) + return None, None, None, None, None, None, None, None + if aot != "dcim.interface": + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm} iface_id={aoid}")) + return None, aoid, None, None, None, None, None, None, None + + # Step 2: Interface -> Device (shallow) + iface, code2 = http_get_json(f"{base}/api/dcim/interfaces/{aoid}/", params={"fields": "device"}, headers=h) + if code2 == 400: + iface, code2 = http_get_json(f"{base}/api/dcim/interfaces/{aoid}/", headers=h) + if code2 != 200 or not iface: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}")) + return None, aoid, None, None, None, None, None, None, None + + dev = iface.get("device") or {} + host = dev.get("name") or dev.get("display") + dev_id = dev.get("id") + if not host or dev_id is None: + asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface has no device iface_id={aoid}")) + return None, aoid, None, None, None, None, None, None, None + + # Step 3: Device detail (single fetch for custom_fields.*) + device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", params={"fields": "custom_fields"}, headers=h) + if code3 == 400: + device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h) + if code3 != 200 or not device: + # treat as "no extra info" + return host, aoid, dev_id, None, None, None, None, None + + cf = device.get("custom_fields") or {} + action_next = cf.get("action_next") + action_last = cf.get("action_last") + action_next_timestamp = cf.get("action_next_timestamp") + action_state = cf.get("action_state") + sot_ts = cf.get("sot_ts") + + # Step 4: populate cache (only on full success) + if NB_LOOKUP_CACHE_TTL > 0: + try: + NB_LOOKUP_CACHE[mac_norm] = ( + monotonic() + NB_LOOKUP_CACHE_TTL, + ( + host, + aoid, + dev_id, + action_next, + action_last, + action_next_timestamp, + action_state, + sot_ts, + ), + ) + except Exception: + pass + + return host, aoid, dev_id, action_next, action_last, action_next_timestamp, action_state, sot_ts + + +# ========================= +# Main +# ========================= +async def main(): + args = parse_args() + ssl_ctx = make_ssl_context(args) + + print_lock = asyncio.Lock() + event_log_lock = asyncio.Lock() + + async def log_status(s: str): + async with print_lock: + print(s, file=sys.stderr, flush=True) + + async def log_event(event: Dict[str, Any]): + try: + async with event_log_lock: + append_jsonl(EVENT_LOG_PATH, event) + except Exception as e: + await log_status(f"[{ts()}] event_log write error path={EVENT_LOG_PATH!r} err={e!r}") + + DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) + recent_payloads: Dict[bytes, float] = {} + + async def disconnected_cb(): + await log_status(f"[{ts()}] Disconnected from NATS.") + + async def reconnected_cb(): + await log_status(f"[{ts()}] Reconnected to NATS.") + + async def error_cb(e): + await log_status(f"[{ts()}] Error: {e!r}") + + async def closed_cb(): + await log_status(f"[{ts()}] Connection closed.") + + nc = await nats.connect( + servers=args.servers, + name=args.name, + allow_reconnect=True, + reconnect_time_wait=2, + max_reconnect_attempts=-1, + disconnected_cb=disconnected_cb, + reconnected_cb=reconnected_cb, + error_cb=error_cb, + closed_cb=closed_cb, + user_credentials=args.creds if args.creds else None, + user=args.user, + password=args.password, + token=args.token, + tls=ssl_ctx, + ) + + async def message_handler(msg: nats.aio.msg.Msg): + t_start = monotonic() + payload = msg.data + event = { + "ts": ts_iso(), + "event_type": "registration_attempt", + "subject": msg.subject, + "product": "-", + "mac": "-", + "mac_norm": None, + "fw": "-", + "device_hostname": None, + "iface_id": None, + "nb_lookup_result": "not_attempted", + "ignored": False, + "ignore_reason": None, + "action_next": None, + "action_last": None, + "action_state": None, + "decision": "received", + "decision_reason": None, + "publish_attempted": False, + "publish_result": "not_attempted", + "routing_key": None, + "task_name": None, + "delay_ms": None, + "netbox_ms": None, + "total_ms": None, + "nb_problems": None, + "reg_age_s": None, + } + + # strict payload dedupe + digest = hashlib.blake2b(payload, digest_size=16).digest() + nowm = monotonic() + exp = recent_payloads.get(digest) + if exp and exp > nowm: + return + recent_payloads[digest] = nowm + DEDUPE_TTL + if len(recent_payloads) > 4096: + cutoff = nowm + for k in list(recent_payloads.keys()): + if recent_payloads[k] <= cutoff: + recent_payloads.pop(k, None) + + product = mac = fw = "-" + try: + text = payload.decode("utf-8", errors="replace") + obj = json.loads(text) + product, mac, fw = extract_fields(obj) + event["reg_age_s"] = extract_registration_age_s(obj) + except Exception: + pass + + event["product"] = product + event["mac"] = mac + event["fw"] = fw + + mac_norm = normalize_mac(mac) + event["mac_norm"] = mac_norm + if mac_norm in IGNORE_MACS: + event["ignored"] = True + event["ignore_reason"] = "mac_ignore_list" + event["decision"] = "ignored" + event["decision_reason"] = "MAC is in local ignore list" + event["total_ms"] = round((monotonic() - t_start) * 1000, 1) + async with NB_PROBLEM_LOCK: + event["nb_problems"] = NB_PROBLEM_COUNTER + await log_event(event) + async with print_lock: + print( + f"[{ts()}] ignoring registration for mac={mac}", + file=sys.stdout, + flush=True, + ) + return + + host_suffix = "" + action_suffix = "" # kept; not used + bell_prefix = "" # ASCII BEL when action_next present (3x) + netbox_time_ms = 0.0 + + if product == "fox100": + nb_start = time.perf_counter() + try: + host, iface_id, dev_id, action_next, action_last, action_next_timestamp, action_state, sot_ts = nb_lookup_device_by_mac( + mac=mac, log_status=log_status + ) + event["iface_id"] = iface_id + event["device_hostname"] = host + event["action_next"] = action_next + event["action_last"] = action_last + event["action_state"] = action_state + if host: + event["nb_lookup_result"] = "resolved" + host_suffix = f" device_hostname={host}" + elif iface_id: + event["nb_lookup_result"] = "iface_only" + else: + event["nb_lookup_result"] = "not_found" + if iface_id and not host: + host_suffix += f" iface_id={iface_id}" + + if host: + # Gate on action_state: allow only "" or "ready" + action_state_str = "" + try: + if action_state is None: + action_state_str = "" + elif isinstance(action_state, str): + action_state_str = action_state.strip() + else: + action_state_str = str(action_state).strip() + except Exception: + action_state_str = "" + + if action_state_str not in ("", "ready", "done"): + event["action_state"] = action_state_str + event["decision"] = "blocked_action_state" + event["decision_reason"] = "device is not ready because of action_state" + event["netbox_ms"] = round((time.perf_counter() - nb_start) * 1000, 1) + event["total_ms"] = round((monotonic() - t_start) * 1000, 1) + async with NB_PROBLEM_LOCK: + event["nb_problems"] = NB_PROBLEM_COUNTER + await log_event(event) + async with print_lock: + print( + f"[{ts()}] device is not ready because of action_state device_hostname={host} action_state={action_state_str}", + file=sys.stdout, + flush=True, + ) + return + + # Determine if action_next is present (non-empty string, or any truthy value) + has_action_next = False + action_next_str = None + try: + if isinstance(action_next, str): + action_next_str = action_next.strip() + has_action_next = len(action_next_str) > 0 + else: + has_action_next = bool(action_next) + if has_action_next: + action_next_str = str(action_next) + except Exception: + has_action_next = False + action_next_str = None + + if not has_action_next: + event["decision"] = "no_action" + event["decision_reason"] = "no action_next for device_hostname" + # User-requested behavior: if no action -> just shoot a message to stdout and we're ok + async with print_lock: + print(f"[{ts()}] no action_next for device_hostname={host}", file=sys.stdout, flush=True) + else: + # Compare action_last with action_next (strings) + action_last_str = None + try: + if isinstance(action_last, str): + action_last_str = action_last.strip() + elif action_last is None: + action_last_str = None + else: + action_last_str = str(action_last) + except Exception: + action_last_str = None + + now_epoch = int(time.time()) + + # Gate: if action_next is posture_analyzer and sot_ts is recent, skip sending task + skip_due_sot = False + event["action_next"] = action_next_str + event["action_last"] = action_last_str + if action_next_str == posture_analyzer: + try: + if isinstance(sot_ts, str): + _st = sot_ts.strip() + if _st: + _dt = datetime.strptime(_st, "%d%m%y-%H%M%S").replace(tzinfo=timezone(timedelta(hours=2))) + _age = now_epoch - int(_dt.timestamp()) + if _age >= 0 and _age < sot_timeout: + skip_due_sot = True + event["decision"] = "cooldown_sot" + event["decision_reason"] = f"sot_ts is recent age_s={_age}" + async with print_lock: + print( + f"[{ts()}] skip action_next because sot_ts is recent device_hostname={host} task={action_next_str} age_s={_age} sot_ts={_st}", + file=sys.stdout, + flush=True, + ) + except Exception: + skip_due_sot = False + + # If action_last == action_next, apply cooldown based on action_next_timestamp (600s) + if skip_due_sot: + pass + elif action_last_str == action_next_str: + allow_repeat = True + try: + if action_next_timestamp is None: + allow_repeat = True + elif isinstance(action_next_timestamp, (int, float)): + allow_repeat = (now_epoch - int(action_next_timestamp)) >= 600 + elif isinstance(action_next_timestamp, str): + allow_repeat = (now_epoch - int(action_next_timestamp.strip())) >= 600 + else: + allow_repeat = True + except Exception: + allow_repeat = True + + if not allow_repeat: + event["decision"] = "cooldown" + event["decision_reason"] = "action_next cooldown active" + async with print_lock: + print( + f"[{ts()}] cooldown action_next for device_hostname={host} task={action_next_str}", + file=sys.stdout, + flush=True, + ) + else: + # Publish task_name=action_next + effective_delay_ms = RMQ_DELAY_MS + target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK + rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish" + + event["publish_attempted"] = True + event["publish_result"] = "attempted" + event["routing_key"] = RMQ_ROUTING_KEY + event["task_name"] = action_next_str + payload_obj = { + "inscope_device": host, + "task_name": action_next_str, + } + + payload_raw = json.dumps(payload_obj, separators=(",", ":"), ensure_ascii=False) + + rmq_body = { + "properties": { + "content_type": "application/json" + }, + "routing_key": RMQ_ROUTING_KEY, + "payload": payload_raw, + "payload_encoding": "string", + } + + if effective_delay_ms > 0: + rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)} + event["delay_ms"] = int(effective_delay_ms) + + # ---- SURGICAL FIX (bell must sound during payload_raw line) ---- + async with print_lock: + sys.stdout.write("\a" * 3) + sys.stdout.flush() + # ------------------------------------------------------------- + + await log_status( + f"[{ts()}] ok, here i will execute\n" + f" url: {rmq_url}\n" + f" routing_key: {RMQ_ROUTING_KEY}\n" + f" payload_raw: {payload_raw}\n" + f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}" + ) + + resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT) + if code != 200: + event["decision"] = "publish_failed" + event["decision_reason"] = f"rmq publish http={code}" + event["publish_result"] = "failed" + await log_status(f"[{ts()}] rmq: publish http={code} device_hostname={host}") + else: + routed = False + try: + routed = bool((resp or {}).get("routed", False)) + except Exception: + routed = False + + # ---- SURGICAL FIX ---- + # For delayed publishes (effective_delay_ms > 0), routed may be false but the message is accepted. + publish_ok = True + if effective_delay_ms <= 0 and not routed: + publish_ok = False + event["decision"] = "publish_failed" + event["decision_reason"] = "rmq immediate publish routed=false" + event["publish_result"] = "failed" + await log_status(f"[{ts()}] rmq: publish immediate routed=false device_hostname={host}") + # ---------------------- + + if publish_ok: + event["decision"] = "scheduled" + event["decision_reason"] = "publish sent to rabbitmq" + event["publish_result"] = "sent" + # On success: set action_last and action_next_timestamp and action_state + try: + base = NB_URL.rstrip("/") + nb_headers = { + "Accept": "application/json", + "Authorization": f"Token {NB_TOKEN}", + } + patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": str(now_epoch), "action_state": "started"}} + _, pcode = http_patch_json( + f"{base}/api/dcim/devices/{dev_id}/", + patch_body, + headers=nb_headers, + timeout=NB_TIMEOUT, + ) + if pcode != 200: + await log_status(f"[{ts()}] nb: action_last/timestamp patch http={pcode} device_hostname={host} dev_id={dev_id}") + except Exception as e: + await log_status(f"[{ts()}] nb: action_last/timestamp patch error device_hostname={host!r} dev_id={dev_id!r} err={e!r}") + + bell_prefix = "\a" * 3 + else: + # action_last != action_next -> publish + effective_delay_ms = RMQ_DELAY_MS + target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK + rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish" + + event["publish_attempted"] = True + event["publish_result"] = "attempted" + event["routing_key"] = RMQ_ROUTING_KEY + event["task_name"] = action_next_str + payload_obj = { + "inscope_device": host, + "task_name": action_next_str, + } + + payload_raw = json.dumps(payload_obj, separators=(",", ":"), ensure_ascii=False) + + rmq_body = { + "properties": { + "content_type": "application/json" + }, + "routing_key": RMQ_ROUTING_KEY, + "payload": payload_raw, + "payload_encoding": "string", + } + + if effective_delay_ms > 0: + rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)} + event["delay_ms"] = int(effective_delay_ms) + + # ---- SURGICAL FIX (bell must sound during payload_raw line) ---- + async with print_lock: + sys.stdout.write("\a" * 3) + sys.stdout.flush() + # ------------------------------------------------------------- + + await log_status( + f"[{ts()}] ok, here i will execute\n" + f" url: {rmq_url}\n" + f" routing_key: {RMQ_ROUTING_KEY}\n" + f" payload_raw: {payload_raw}\n" + f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}" + ) + + resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT) + if code != 200: + event["decision"] = "publish_failed" + event["decision_reason"] = f"rmq publish http={code}" + event["publish_result"] = "failed" + await log_status(f"[{ts()}] rmq: publish http={code} device_hostname={host}") + else: + routed = False + try: + routed = bool((resp or {}).get("routed", False)) + except Exception: + routed = False + + # ---- SURGICAL FIX ---- + publish_ok = True + if effective_delay_ms <= 0 and not routed: + publish_ok = False + event["decision"] = "publish_failed" + event["decision_reason"] = "rmq immediate publish routed=false" + event["publish_result"] = "failed" + await log_status(f"[{ts()}] rmq: publish immediate routed=false device_hostname={host}") + # ---------------------- + + if publish_ok: + event["decision"] = "scheduled" + event["decision_reason"] = "publish sent to rabbitmq" + event["publish_result"] = "sent" + now_epoch = int(time.time()) + try: + base = NB_URL.rstrip("/") + nb_headers = { + "Accept": "application/json", + "Authorization": f"Token {NB_TOKEN}", + } + patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": str(now_epoch), "action_state": "started"}} + _, pcode = http_patch_json( + f"{base}/api/dcim/devices/{dev_id}/", + patch_body, + headers=nb_headers, + timeout=NB_TIMEOUT, + ) + if pcode != 200: + await log_status(f"[{ts()}] nb: action_last/timestamp patch http={pcode} device_hostname={host} dev_id={dev_id}") + except Exception as e: + await log_status(f"[{ts()}] nb: action_last/timestamp patch error device_hostname={host!r} dev_id={dev_id!r} err={e!r}") + + bell_prefix = "" + + except Exception as e: + event["decision"] = "error" + event["decision_reason"] = f"nb unexpected error {e!r}" + event["nb_lookup_result"] = "error" + 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 + + total_ms = (monotonic() - t_start) * 1000 + + async with NB_PROBLEM_LOCK: + nb_problems_snapshot = NB_PROBLEM_COUNTER + + event["netbox_ms"] = round(netbox_time_ms, 1) if product == "fox100" else None + event["total_ms"] = round(total_ms, 1) + event["nb_problems"] = nb_problems_snapshot + if event["decision"] == "received": + if product != "fox100": + event["decision"] = "unsupported_product" + event["decision_reason"] = "product is not handled by fox100 logic" + elif event["nb_lookup_result"] == "resolved": + event["decision"] = "lookup_only" + event["decision_reason"] = "registration processed without further action" + elif event["nb_lookup_result"] == "iface_only": + event["decision"] = "lookup_partial" + event["decision_reason"] = "mac resolved to interface only" + elif event["nb_lookup_result"] == "not_found": + event["decision"] = "lookup_failed" + event["decision_reason"] = "mac not found in netbox" + await log_event(event) + + reg_age_s = event.get("reg_age_s") + reg_age_suffix = f" reg_age_s={reg_age_s}" if reg_age_s is not None else " reg_age_s=na" + line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}{action_suffix}{reg_age_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(bell_prefix + line + "\n") + sys.stdout.flush() + + 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={args.queue or '-'}) via {args.servers} | nb=on" + ) + + stop_event = asyncio.Event() + + def handle_signal(*_): + asyncio.create_task(log_status(f"[{ts()}] Received stop signal, draining...")) + stop_event.set() + + loop = asyncio.get_running_loop() + for s in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(s, handle_signal) + except NotImplementedError: + signal.signal(s, lambda *_: handle_signal()) + + await stop_event.wait() + await nc.drain() + await nc.close() + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass \ No newline at end of file diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 4db4ffe..d7585a9 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -518,8 +518,9 @@ async def main(): tls=ssl_ctx, ) - async def message_handler(msg: nats.aio.msg.Msg): + async def message_handler(msg: nats.aio.msg.Msg, queued_monotonic: Optional[float] = None, queued_iso: Optional[str] = None): t_start = monotonic() + local_queue_depth = processing_queue.qsize() payload = msg.data event = { "ts": ts_iso(), @@ -548,7 +549,9 @@ async def main(): "total_ms": None, "nb_problems": None, "reg_age_s": None, + "local_queue_depth": None, } + event["local_queue_depth"] = local_queue_depth # strict payload dedupe digest = hashlib.blake2b(payload, digest_size=16).digest() @@ -953,7 +956,8 @@ async def main(): reg_age_s = event.get("reg_age_s") reg_age_suffix = f" reg_age_s={reg_age_s}" if reg_age_s is not None else " reg_age_s=na" - line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}{action_suffix}{reg_age_suffix}" + queue_suffix = f" local_queue={local_queue_depth}" + line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}{action_suffix}{reg_age_suffix}{queue_suffix}" if product == "fox100": line += f" netbox_ms={netbox_time_ms:.1f} total_ms={total_ms:.1f}" if args.include_subject: @@ -963,10 +967,28 @@ async def main(): sys.stdout.write(bell_prefix + line + "\n") sys.stdout.flush() + processing_queue: asyncio.Queue = asyncio.Queue() + + async def enqueue_message(msg: nats.aio.msg.Msg): + await processing_queue.put((msg, monotonic(), ts_iso())) + + async def processing_worker(): + while True: + item = await processing_queue.get() + try: + if item is None: + return + qmsg, queued_monotonic, queued_iso = item + await message_handler(qmsg, queued_monotonic=queued_monotonic, queued_iso=queued_iso) + finally: + processing_queue.task_done() + + worker_task = asyncio.create_task(processing_worker()) + if args.queue: - await nc.subscribe(args.subject, queue=args.queue, cb=message_handler) + await nc.subscribe(args.subject, queue=args.queue, cb=enqueue_message) else: - await nc.subscribe(args.subject, cb=message_handler) + await nc.subscribe(args.subject, cb=enqueue_message) await log_status( f"[{ts()}] Listening on subject '{args.subject}' (queue={args.queue or '-'}) via {args.servers} | nb=on" @@ -987,6 +1009,9 @@ async def main(): await stop_event.wait() await nc.drain() + await processing_queue.join() + await processing_queue.put(None) + await worker_task await nc.close()