From 7361a85f027217916c0c75ba5776732a15f040df Mon Sep 17 00:00:00 2001 From: pavel Date: Mon, 9 Mar 2026 08:23:49 +0200 Subject: [PATCH] 0823 --- files/nats_registration_listener.py | 150 +++++++++++++++++++++++++--- 1 file changed, 138 insertions(+), 12 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 67ec4d3..f3f7851 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -30,6 +30,7 @@ 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 @@ -72,19 +73,13 @@ PRODUCT_TAG_SLUG = { # "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:00:D5:84", - "44:D1:FA:7E:3F:65", - "D0:6C:37:00:E6:D8", - "C4:93:00:4E:92:7C", -] - # 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 @@ -128,6 +123,17 @@ 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"] @@ -167,9 +173,6 @@ def normalize_mac(mac: str) -> Optional[str]: 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)}" @@ -376,11 +379,19 @@ async def main(): 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] = {} @@ -416,6 +427,33 @@ async def main(): 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": "-", + "host": 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, + } # strict payload dedupe digest = hashlib.blake2b(payload, digest_size=16).digest() @@ -438,8 +476,21 @@ async def main(): 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}", @@ -459,8 +510,18 @@ async def main(): host, iface_id, dev_id, status_val, tag_slugs, 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["host"] = 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" host={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}" @@ -478,6 +539,14 @@ async def main(): 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 host={host} action_state={action_state_str}", @@ -502,6 +571,8 @@ async def main(): action_next_str = None if not has_action_next: + event["decision"] = "no_action" + event["decision_reason"] = "no action_next for host" # 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 host={host}", file=sys.stdout, flush=True) @@ -522,6 +593,8 @@ async def main(): # 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): @@ -531,6 +604,8 @@ async def main(): _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 host={host} task={action_next_str} age_s={_age} sot_ts={_st}", @@ -558,6 +633,8 @@ async def main(): 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 host={host} task={action_next_str}", @@ -570,6 +647,10 @@ async def main(): 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, @@ -588,6 +669,7 @@ async def main(): 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: @@ -605,6 +687,9 @@ async def main(): 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} host={host}") else: routed = False @@ -618,10 +703,16 @@ async def main(): 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 host={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("/") @@ -648,6 +739,10 @@ async def main(): 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, @@ -666,6 +761,7 @@ async def main(): 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: @@ -683,6 +779,9 @@ async def main(): 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} host={host}") else: routed = False @@ -695,10 +794,16 @@ async def main(): 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 host={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("/") @@ -721,6 +826,9 @@ async def main(): 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 @@ -729,6 +837,24 @@ async def main(): 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) + line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}{action_suffix}" if product == "fox100": line += f" netbox_ms={netbox_time_ms:.1f} total_ms={total_ms:.1f}"