diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index cc0bbcd..a190beb 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -6,10 +6,11 @@ NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing - 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 " - (and publish to RabbitMQ) - * After successful publish (HTTP 200 + routed true): copy action_next -> action_last in NetBox (PATCH) + * 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) """ @@ -214,15 +215,17 @@ 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], Optional[Any], Optional[Any] +]: """ - Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next) + Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next, action_last, action_next_timestamp) - Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail). - - If device detail fetch fails, returns host/id with status/tags/action_next as None. + - If device detail fetch fails, returns host/id with status/tags/custom_fields as None. """ mac_norm = normalize_mac(mac) if not mac_norm: - return None, None, None, None, None, None + return None, None, None, None, None, None, None, None base = NB_URL.rstrip("/") h = { @@ -235,12 +238,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, None + 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 + return None, None, None, None, None, None, None, None rec = results[0] aot = (rec.get("assigned_object_type") or "").strip() @@ -251,29 +254,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, None + 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 + return None, aoid, None, None, 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, None + return None, aoid, 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 + return None, aoid, None, None, None, None, None, None - # Step 3: Device detail (single fetch for status, tags, custom_fields.action_next) + # Step 3: Device detail (single fetch for status, tags, custom_fields.*) 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 + return host, aoid, dev_id, None, None, None, None, None status_val = ((device.get("status") or {}).get("value")) or None tags = device.get("tags") or [] @@ -285,8 +288,10 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option 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") - return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next + return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next, action_last, action_next_timestamp # ========================= @@ -367,7 +372,9 @@ async def main(): if product == "fox100": nb_start = time.perf_counter() try: - host, iface_id, dev_id, status_val, tag_slugs, action_next = nb_lookup_device_by_mac(mac=mac, log_status=log_status) + host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp = nb_lookup_device_by_mac( + mac=mac, log_status=log_status + ) if host: host_suffix = f" host={host}" if iface_id and not host: @@ -394,63 +401,162 @@ async def main(): 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" - - 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", - } - - # 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)}" - ) - - # Publish - 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}") - else: - routed = False - try: - routed = bool((resp or {}).get("routed", False)) - except Exception: - routed = False - if not routed: - await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + # 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: - # Copy action_next -> action_last in NetBox (surgical) - try: - base = NB_URL.rstrip("/") - nb_headers = { - "Accept": "application/json", - "Authorization": f"Token {NB_TOKEN}", - } - patch_body = {"custom_fields": {"action_last": action_next_str}} - _, 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 patch http={pcode} host={host} dev_id={dev_id}") - except Exception as e: - await log_status(f"[{ts()}] nb: action_last patch error host={host!r} dev_id={dev_id!r} err={e!r}") + action_last_str = str(action_last) + except Exception: + action_last_str = None - # Bell behavior: ring 3x BEL when action_next present - bell_prefix = "\a" * 3 + now_epoch = int(time.time()) + + # If action_last == action_next, apply cooldown based on action_next_timestamp (600s) + if 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: + async with print_lock: + print( + f"[{ts()}] cooldown action_next for host={host} task={action_next_str}", + file=sys.stdout, + flush=True, + ) + else: + # Publish task_name=action_next + rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish" + + 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", + } + + 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: + await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}") + else: + routed = False + try: + routed = bool((resp or {}).get("routed", False)) + except Exception: + routed = False + if not routed: + await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + else: + # On success: set action_last and action_next_timestamp + 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": now_epoch}} + _, 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} host={host} dev_id={dev_id}") + except Exception as e: + await log_status(f"[{ts()}] nb: action_last/timestamp patch error host={host!r} dev_id={dev_id!r} err={e!r}") + + bell_prefix = "\a" * 3 + else: + # action_last != action_next -> publish immediately + rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish" + + 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", + } + + 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: + await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}") + else: + routed = False + try: + routed = bool((resp or {}).get("routed", False)) + except Exception: + routed = False + if not routed: + await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") + else: + 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": now_epoch}} + _, 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} host={host} dev_id={dev_id}") + except Exception as e: + await log_status(f"[{ts()}] nb: action_last/timestamp patch error host={host!r} dev_id={dev_id!r} err={e!r}") + + bell_prefix = "\a" * 3 except Exception as e: await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")