From f2600d2bb6a849e2e2f4a441951243ed5bd0a631 Mon Sep 17 00:00:00 2001 From: pavel Date: Wed, 5 Nov 2025 20:38:35 +0200 Subject: [PATCH] 20:38 --- files/nats_registration_listener.py | 88 ++++++++++++++++++----------- 1 file changed, 54 insertions(+), 34 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 1bb0e26..eb5dcaf 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,9 +1,12 @@ #!/usr/bin/env python3 """ -NATS Registration Listener (fox100 + NetBox hostname/tag lookup + timing + problem counter) +NATS Registration Listener (fox100 + NetBox hostname/upgrade_cmd lookup + timing + problem counter) ------------------------------------------------------------------------------------------- -- One device GET (status + tags), no duplicate fetch -- On tag match (and status=active), prepend ASCII BEL to stdout line to alert +- One device GET (status + tags + custom_fields.upgrade_cmd), no duplicate fetch +- If fox100 and upgrade_cmd present -> prepend 3x ASCII BEL to stdout line +- Publish logic for fox100: + * upgrade_cmd empty/absent -> publish with task_name="ot-updater" + * upgrade_cmd present -> publish with task_name="sot-updater-upgradecmd" (+ 3 BELs) - Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting """ @@ -22,7 +25,7 @@ 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 # NEW: for RabbitMQ Basic Auth +import base64 # for RabbitMQ Basic Auth import nats @@ -46,7 +49,7 @@ RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) RMQ_ROUTING_KEY = "deviceconfig" RMQ_TIMEOUT = 3.0 -# Product -> Tag slug mapping (future-proof for fox200 later) +# Product -> Tag slug mapping (kept unchanged, though not used now) PRODUCT_TAG_SLUG = { "fox100": "fox100-auto-upgrade-latest", # "fox200": "fox200-auto-upgrade-latest", @@ -156,7 +159,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op return None, 597 -# NEW: RabbitMQ management API POST helper (basic auth; JSON in/out) +# 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"} @@ -187,15 +190,15 @@ 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]]]: +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) + Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, upgrade_cmd) - Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail). - - If device detail fetch fails, returns host/id with status/tags as None (treated as "no tag info"). + - If device detail fetch fails, returns host/id with status/tags/upgrade_cmd as None. """ mac_norm = normalize_mac(mac) if not mac_norm: - return None, None, None, None, None + return None, None, None, None, None, None base = NB_URL.rstrip("/") h = { @@ -208,12 +211,12 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option 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 + return 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 + return None, None, None, None, None, None rec = results[0] aot = (rec.get("assigned_object_type") or "").strip() @@ -224,29 +227,29 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option 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 + return 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 + return None, aoid, None, None, None, None # Step 2: Interface -> Device (shallow) 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 + return None, aoid, 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 + return None, aoid, None, None, None, None - # Step 3: Device detail (single fetch for BOTH status and tags) + # Step 3: Device detail (single fetch for status, tags, custom_fields.upgrade_cmd) device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h) if code3 != 200 or not device: - # treat as "no tag info" (no problem increment) - return host, aoid, dev_id, None, None + # treat as "no extra info" + return host, aoid, dev_id, None, None, None status_val = ((device.get("status") or {}).get("value")) or None tags = device.get("tags") or [] @@ -256,7 +259,10 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option if isinstance(slug, str): tag_slugs.add(slug) - return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs + cf = device.get("custom_fields") or {} + upgrade_cmd = cf.get("upgrade_cmd") + + return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, upgrade_cmd # ========================= @@ -330,30 +336,40 @@ async def main(): pass host_suffix = "" - action_suffix = "" - bell_prefix = "" # ASCII BEL when we have a tag match + action_suffix = "" # kept; no longer used for tag action + bell_prefix = "" # ASCII BEL when upgrade_cmd present (3x) netbox_time_ms = 0.0 - desired_slug = PRODUCT_TAG_SLUG.get(product) + # NOTE: no tag check anymore; behavior depends on upgrade_cmd only if product == "fox100": nb_start = time.perf_counter() try: - host, iface_id, dev_id, status_val, tag_slugs = nb_lookup_device_by_mac(mac=mac, log_status=log_status) + host, iface_id, dev_id, status_val, tag_slugs, upgrade_cmd = nb_lookup_device_by_mac(mac=mac, log_status=log_status) if host: host_suffix = f" host={host}" if iface_id and not host: host_suffix += f" iface_id={iface_id}" - # Tag-based action (unchanged) - if dev_id is not None and status_val == "active" and isinstance(desired_slug, str) and isinstance(tag_slugs, set): - if desired_slug in tag_slugs: - bell_prefix = "\a" * 5 - action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade" - - # NEW: Publish immediate RMQ message like rmq-ikeja-pub3.sh (no delay) + # NEW LOGIC: + # - If host known: + # * upgrade_cmd empty/absent -> publish with task_name="ot-updater" + # * upgrade_cmd present -> publish with task_name="sot-updater-upgradecmd" and ring 3 BELs if host: try: rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish" + + has_upgrade_cmd = False + try: + # consider non-empty string or any truthy value as "present" + if isinstance(upgrade_cmd, str): + has_upgrade_cmd = len(upgrade_cmd.strip()) > 0 + else: + has_upgrade_cmd = bool(upgrade_cmd) + except Exception: + has_upgrade_cmd = False + + task_name = "sot-updater-upgradecmd" if has_upgrade_cmd else "ot-updater" + rmq_body = { "properties": { "content_type": "application/json" @@ -361,7 +377,7 @@ async def main(): "routing_key": RMQ_ROUTING_KEY, "payload": json.dumps({ "inscope_device": host, - "task_name": "sot-updater-upgradecmd", + "task_name": task_name, }), "payload_encoding": "string", } @@ -369,7 +385,6 @@ async def main(): if code != 200: await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}") else: - # if response JSON has 'routed' false, note it (rmq-ikeja-pub3.sh warns in that case) routed = False try: routed = bool((resp or {}).get("routed", False)) @@ -377,6 +392,11 @@ async def main(): routed = False if not routed: await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + + # Bell behavior: ring 3x BEL only if upgrade_cmd present + if has_upgrade_cmd: + bell_prefix = "\a" * 3 + except Exception as e: await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}") @@ -396,7 +416,7 @@ async def main(): line += f" subject={msg.subject}" async with print_lock: - # Prepend BEL only when we had a tag match + # Prepend BEL only when we had upgrade_cmd present (3x) sys.stdout.write(bell_prefix + line + "\n") sys.stdout.flush()