pavel/nats-scheduler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0957

Browse Source
This commit is contained in:
pavel
2026-02-15 09:57:13 +02:00
parent ef4820661c
commit d3c3a32e7c
1 changed files with 27 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
files/nats_registration_listener.py
View File

@@ -58,6 +58,10 @@ 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)
# Product -> Tag slug mapping (kept unchanged, though not used now)
PRODUCT_TAG_SLUG = {
"fox100": "fox100-auto-upgrade-latest",
@@ -424,6 +428,7 @@ async def main():
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:
@@ -464,8 +469,15 @@ async def main():
flush=True,
)
else:
# Publish task_name=action_next
target_exchange = RMQ_EXCHANGE_DELAYED if RMQ_DELAY_MS > 0 else RMQ_EXCHANGE_WORK
# ---- Delay override logic (user-requested)
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)
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"
payload_obj = {
@@ -484,8 +496,8 @@ async def main():
"payload_encoding": "string",
}
if RMQ_DELAY_MS > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(RMQ_DELAY_MS)}
if effective_delay_ms > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)}
await log_status(
f"[{ts()}] ok, here i will execute\n"
@@ -504,11 +516,7 @@ async def main():
routed = bool((resp or {}).get("routed", False))
except Exception:
routed = False
# FIX: delayed exchange accepts messages even if routed=false
publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed
if not publish_ok:
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 and action_state
@@ -538,8 +546,13 @@ async def main():
bell_prefix = "\a" * 3
else:
# action_last != action_next -> publish immediately
target_exchange = RMQ_EXCHANGE_DELAYED if RMQ_DELAY_MS > 0 else RMQ_EXCHANGE_WORK
# action_last != action_next -> publish
# ---- Delay override logic (user-requested)
effective_delay_ms = RMQ_DELAY_MS
if action_next_str == POSTURE_SCAN_SOT:
effective_delay_ms = int(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"
payload_obj = {
@@ -558,8 +571,8 @@ async def main():
"payload_encoding": "string",
}
if RMQ_DELAY_MS > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(RMQ_DELAY_MS)}
if effective_delay_ms > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)}
await log_status(
f"[{ts()}] ok, here i will execute\n"
@@ -578,11 +591,7 @@ async def main():
routed = bool((resp or {}).get("routed", False))
except Exception:
routed = False
# FIX: delayed exchange accepts messages even if routed=false
publish_ok = True if target_exchange == RMQ_EXCHANGE_DELAYED else routed
if not publish_ok:
if not routed:
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
else:
now_epoch = int(time.time())
@@ -663,4 +672,3 @@ if __name__ == "__main__":
asyncio.run(main())
except KeyboardInterrupt:
pass