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

1026

Browse Source
This commit is contained in:
pavel
2026-02-15 10:26:21 +02:00
parent d3c3a32e7c
commit 7cb47d2188
1 changed files with 33 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
files/nats_registration_listener.py
View File

@@ -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,