From 7cb47d21885249afe40f5e5748de0ca0495fb420 Mon Sep 17 00:00:00 2001 From: pavel Date: Sun, 15 Feb 2026 10:26:21 +0200 Subject: [PATCH] 1026 --- files/nats_registration_listener.py | 68 ++++++++++++++--------------- 1 file changed, 33 insertions(+), 35 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 247a186..06772f4 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -49,7 +49,7 @@ RMQ_PORT = 15672 RMQ_USER = "admin" RMQ_PASS = "change_me" RMQ_VHOST = "app" -RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) +RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate) RMQ_EXCHANGE_DELAYED = "deviceconfig.delayed" # delayed exchange (x-delayed-message) RMQ_ROUTING_KEY = "deviceconfig" RMQ_TIMEOUT = 3.0 @@ -58,9 +58,11 @@ RMQ_TIMEOUT = 3.0 # Example: 600000 = 10 minutes RMQ_DELAY_MS = 15000 -# ---- Posture scan special-case (user-requested variables) -POSTURE_SCAN_SOT = "sot-updater-wrapper" # "posture-scan-sot" -POSTURE_SOT_WAITTIME_MS = 30000 # "posture-sot-waittime" (ms) +# ---- Per-task delay overrides (human-editable) +# When action_next equals this task name, special delay rules apply. +POSTURE_SCAN_SOT = "sot-updater-wrapper" +# Delay to use when re-scheduling POSTURE_SCAN_SOT after a posture scan already happened. +POSTURE_SOT_WAITTIME_MS = 30000 # Product -> Tag slug mapping (kept unchanged, though not used now) PRODUCT_TAG_SLUG = { @@ -391,7 +393,7 @@ async def main(): host_suffix += f" iface_id={iface_id}" if host: - # Gate on action_state: allow only "" or "ready" or "done" + # Gate on action_state: allow only "" or "ready" action_state_str = "" try: if action_state is None: @@ -469,13 +471,12 @@ async def main(): flush=True, ) else: - # ---- Delay override logic (user-requested) + # Publish task_name=action_next effective_delay_ms = RMQ_DELAY_MS - if action_next_str == POSTURE_SCAN_SOT: - if action_last_str == action_next_str: - effective_delay_ms = 0 - else: - effective_delay_ms = int(POSTURE_SOT_WAITTIME_MS) + if action_next_str == POSTURE_SCAN_SOT and action_last_str == action_next_str: + effective_delay_ms = 0 + elif action_next_str == POSTURE_SCAN_SOT and action_last_str != action_next_str and action_last_str not in (None, ""): + effective_delay_ms = POSTURE_SOT_WAITTIME_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" @@ -509,16 +510,20 @@ async def main(): 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}") + await log_status(f"[{ts()}] rmq: publish http={code} host={host}") else: routed = False try: routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - if not routed: + + publish_ok = True + if target_exchange != RMQ_EXCHANGE_DELAYED and not routed: + publish_ok = False await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") - else: + + if publish_ok: # On success: set action_last and action_next_timestamp and action_state try: base = NB_URL.rstrip("/") @@ -526,13 +531,7 @@ async def main(): "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", - } - } + 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, @@ -546,11 +545,12 @@ async def main(): bell_prefix = "\a" * 3 else: - # action_last != action_next -> publish - # ---- Delay override logic (user-requested) + # action_last != action_next -> publish immediately effective_delay_ms = RMQ_DELAY_MS - if action_next_str == POSTURE_SCAN_SOT: - effective_delay_ms = int(POSTURE_SOT_WAITTIME_MS) + if action_next_str == POSTURE_SCAN_SOT and action_last_str == action_next_str: + effective_delay_ms = 0 + elif action_next_str == POSTURE_SCAN_SOT and action_last_str != action_next_str and action_last_str not in (None, ""): + effective_delay_ms = POSTURE_SOT_WAITTIME_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" @@ -584,16 +584,20 @@ async def main(): 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}") + await log_status(f"[{ts()}] rmq: publish http={code} host={host}") else: routed = False try: routed = bool((resp or {}).get("routed", False)) except Exception: routed = False - if not routed: + + publish_ok = True + if target_exchange != RMQ_EXCHANGE_DELAYED and not routed: + publish_ok = False await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}") - else: + + if publish_ok: now_epoch = int(time.time()) try: base = NB_URL.rstrip("/") @@ -601,13 +605,7 @@ async def main(): "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", - } - } + 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,