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

08:35

Browse Source
This commit is contained in:
pavel
2025-11-04 08:35:58 +02:00
parent 5964f6603e
commit 89044dc398
1 changed files with 34 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
files/nats_registration_listener.py
View File

@@ -35,17 +35,16 @@ NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided
NB_TIMEOUT = 3.0 # seconds per HTTP GET NB_TIMEOUT = 3.0 # seconds per HTTP GET
# ========================= # =========================
# RabbitMQ hardcoded config (TTL + DLX path) # RabbitMQ hardcoded config (immediate publish like rmq-ikeja-pub3.sh without delay)
# ========================= # =========================
RMQ_HOST = "10.210.12.2" RMQ_HOST = "10.210.12.2"
RMQ_PORT = 15672 RMQ_PORT = 15672
RMQ_USER = "admin" RMQ_USER = "admin"
RMQ_PASS = "change_me" RMQ_PASS = "change_me"
RMQ_VHOST = "app" RMQ_VHOST = "app"
RMQ_EXCHANGE_HOLDING = "deviceconfig.holding" # publish here with per-message TTL RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate)
RMQ_ROUTING_KEY = "deviceconfig" # routed by DLX to live flow RMQ_ROUTING_KEY = "deviceconfig"
RMQ_TIMEOUT = 3.0 RMQ_TIMEOUT = 3.0
RMQ_DELAY_MS = 10000 # 10 seconds
# Product -> Tag slug mapping (future-proof for fox200 later) # Product -> Tag slug mapping (future-proof for fox200 later)
PRODUCT_TAG_SLUG = { PRODUCT_TAG_SLUG = {
@@ -132,7 +131,7 @@ def normalize_mac(mac: str) -> Optional[str]:
s = mac.strip().lower().replace("-", ":") s = mac.strip().lower().replace("-", ":")
hex_only = "".join(ch for ch in s if ch in "0123456789abcdef") hex_only = "".join(ch for ch in s if ch in "0123456789abcdef")
if len(hex_only) == 12: if len(hex_only) == 12:
return ":".join(hex_only[i:i+2] for i in range(0, 12, 2)) return ":join".replace(":", "").join([":".join(hex_only[i:i+2] for i in range(0, 12, 2))]) # (keeping original behavior; no change)
parts = s.split(":") parts = s.split(":")
if len(parts) == 6 and all(len(p) == 2 and all(c in "0123456789abcdef" for c in p) for p in parts): if len(parts) == 6 and all(len(p) == 2 and all(c in "0123456789abcdef" for c in p) for p in parts):
return s return s
@@ -345,34 +344,42 @@ async def main():
if iface_id and not host: if iface_id and not host:
host_suffix += f" iface_id={iface_id}" host_suffix += f" iface_id={iface_id}"
# Tag-based action # 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 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: if desired_slug in tag_slugs:
# Beep to draw attention
bell_prefix = "\a" * 5 bell_prefix = "\a" * 5
action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade" action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
# NEW: Schedule delayed RabbitMQ message (TTL + DLX; 10s) # NEW: Publish immediate RMQ message like rmq-ikeja-pub3.sh (no delay)
if host: if host:
try:
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
rmq_body = {
"properties": {
"content_type": "application/json"
},
"routing_key": RMQ_ROUTING_KEY,
"payload": json.dumps({
"inscope_device": host,
"task_name": "sot-updater",
}),
"payload_encoding": "string",
}
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:
# if response JSON has 'routed' false, note it (rmq-ikeja-pub3.sh warns in that case)
routed = False
try: try:
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_HOLDING}/publish" routed = bool((resp or {}).get("routed", False))
rmq_body = { except Exception:
"properties": { routed = False
"content_type": "application/json", if not routed:
"expiration": str(RMQ_DELAY_MS), await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
}, except Exception as e:
"routing_key": RMQ_ROUTING_KEY, await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
"payload": json.dumps({
"inscope_device": host,
"task_name": "sot-updater",
}),
"payload_encoding": "string",
}
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 delayed http={code} host={host} delay_ms={RMQ_DELAY_MS}")
except Exception as e:
await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
except Exception as e: except Exception as e:
await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
netbox_time_ms = (time.perf_counter() - nb_start) * 1000 netbox_time_ms = (time.perf_counter() - nb_start) * 1000