From 46ea4359edeb1e132a30074aebfda7bfc7657e86 Mon Sep 17 00:00:00 2001 From: pavel Date: Wed, 11 Feb 2026 14:35:50 +0200 Subject: [PATCH] 1435 --- files/nats_registration_listener.py | 119 +++++++++++++--------------- 1 file changed, 55 insertions(+), 64 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index e850ddd..bec7ae2 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,14 +1,15 @@ #!/usr/bin/env python3 """ NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing + problem counter) ------------------------------------------------------------------------------------------------ +------------------------------------------------------------------------------------------- - One device GET (status + tags + custom_fields.action_next), no duplicate fetch -- If fox100 and action_next present -> prepend 3x ASCII BEL to stdout line -- RabbitMQ publish logic is DISABLED for now: - * We still PREPARE the RabbitMQ message body - * We clearly log: "ok, here i will execute " - * But we DO NOT actually POST it to RabbitMQ (publish call commented out) - 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 is present -> PREPARE RabbitMQ /publish body and log: + "ok, here i will execute " + (but DO NOT actually publish; publish call remains commented out) + * If action_next present -> prepend 3x ASCII BEL to stdout line (kept behavior) """ import argparse @@ -39,7 +40,8 @@ NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided NB_TIMEOUT = 3.0 # seconds per HTTP GET # ========================= -# RabbitMQ hardcoded config (publish currently disabled) +# RabbitMQ hardcoded config (immediate publish like rmq-ikeja-pub3.sh without delay) +# NOTE: publish is currently disabled in message_handler (kept config unchanged). # ========================= RMQ_HOST = "10.210.12.2" RMQ_PORT = 15672 @@ -130,29 +132,19 @@ def extract_fields(obj: Dict[str, Any]): # NetBox lookup (urllib) # ========================= def normalize_mac(mac: str) -> Optional[str]: - """ - Normalize MAC to lowercase colon-separated hex pairs: aa:bb:cc:dd:ee:ff - Accepts common formats: aa-bb-cc-dd-ee-ff, aabbccddeeff, aa:bb:... - """ if not mac or not isinstance(mac, str): return None - s = mac.strip().lower() - - # Strip all non-hex chars to handle aabb.ccdd.eeff, aa-bb-..., etc. + s = mac.strip().lower().replace("-", ":") hex_only = "".join(ch for ch in s if ch in "0123456789abcdef") if len(hex_only) == 12: - return ":".join(hex_only[i:i + 2] for i in range(0, 12, 2)) - - # If it's already colon-separated, validate it strictly. - parts = s.replace("-", ":").split(":") + 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 ":".join(parts) - + return s return None -def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, - timeout: float = NB_TIMEOUT): +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") @@ -171,9 +163,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op # RabbitMQ management API POST helper (basic auth; JSON in/out) -# NOTE: currently not used because publishing is disabled, but left in place for later. -def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, - timeout: float = RMQ_TIMEOUT): +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: @@ -203,9 +193,7 @@ async def nb_problem(log_status, msg: str): await log_status(msg) -def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ - Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any] -]: +def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any]]: """ Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next) - Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail). @@ -223,8 +211,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ } # Step 1: MAC lookup - data, code = http_get_json(f"{base}/api/dcim/mac-addresses/", params={"mac_address": mac_norm, "limit": "2"}, - headers=h) + 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 @@ -239,9 +226,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[ 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 '-'}") - ) + 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}")) @@ -354,11 +339,10 @@ async def main(): pass host_suffix = "" - action_suffix = "" # kept; no longer used for tag action + action_suffix = "" # kept; not used bell_prefix = "" # ASCII BEL when action_next present (3x) netbox_time_ms = 0.0 - # Behavior depends on action_next presence only (for fox100) if product == "fox100": nb_start = time.perf_counter() try: @@ -368,43 +352,56 @@ async def main(): if iface_id and not host: host_suffix += f" iface_id={iface_id}" - # If host known, prepare RabbitMQ message and log it, but DO NOT publish. if host: + # 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: + # 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) + else: + # Prepare RabbitMQ message with task_name taken from action_next (do not keep old task names) rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish" - has_action_next = False - try: - if isinstance(action_next, str): - has_action_next = len(action_next.strip()) > 0 - else: - has_action_next = bool(action_next) - except Exception: - has_action_next = False - - # Task naming: keep it simple and explicit. - task_name = "sot-updater-actionnext" if has_action_next else "sot-updater" - - rmq_payload_obj = { + payload_obj = { "inscope_device": host, - "task_name": task_name, - # IMPORTANT: we are not "executing" it here; we only carry it for later scripts. - "action_next": action_next, + "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": json.dumps(rmq_payload_obj), + "payload": payload_raw, "payload_encoding": "string", } - # (b) Explicit log of what would be executed - await log_status(f"[{ts()}] ok, here i will execute {json.dumps(rmq_body, ensure_ascii=False)} url={rmq_url}") + # Visual, easy-to-read log of what would be executed + 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)}" + ) - # (c) Publishing is disabled for now (leave code in place, commented out) + # Publish disabled (kept in place, commented out) # resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT) # if code != 200: # await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}") @@ -417,12 +414,8 @@ async def main(): # if not routed: # await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") - # Bell behavior: ring 3x BEL only if action_next present - if has_action_next: - bell_prefix = "\a" * 3 - - except Exception as e: - await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}") + # Bell behavior: ring 3x BEL when action_next present + bell_prefix = "\a" * 3 except Exception as e: await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") @@ -440,7 +433,6 @@ async def main(): line += f" subject={msg.subject}" async with print_lock: - # Prepend BEL only when we had action_next present (3x) sys.stdout.write(bell_prefix + line + "\n") sys.stdout.flush() @@ -476,4 +468,3 @@ if __name__ == "__main__": asyncio.run(main()) except KeyboardInterrupt: pass -