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

20:38

Browse Source
This commit is contained in:
pavel
2025-11-05 20:38:35 +02:00
parent f85147e7c0
commit f2600d2bb6
1 changed files with 54 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
files/nats_registration_listener.py
View File

@@ -1,9 +1,12 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (fox100 + NetBox hostname/tag lookup + timing + problem counter)
NATS Registration Listener (fox100 + NetBox hostname/upgrade_cmd lookup + timing + problem counter)
-------------------------------------------------------------------------------------------
- One device GET (status + tags), no duplicate fetch
- On tag match (and status=active), prepend ASCII BEL to stdout line to alert
- One device GET (status + tags + custom_fields.upgrade_cmd), no duplicate fetch
- If fox100 and upgrade_cmd present -> prepend 3x ASCII BEL to stdout line
- Publish logic for fox100:
* upgrade_cmd empty/absent -> publish with task_name="ot-updater"
* upgrade_cmd present -> publish with task_name="sot-updater-upgradecmd" (+ 3 BELs)
- Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting
"""
@@ -22,7 +25,7 @@ from typing import Optional, Dict, Any, Tuple, Set
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import base64 # NEW: for RabbitMQ Basic Auth
import base64 # for RabbitMQ Basic Auth
import nats
@@ -46,7 +49,7 @@ RMQ_EXCHANGE_WORK = "deviceconfig" # direct exchange (immediate)
RMQ_ROUTING_KEY = "deviceconfig"
RMQ_TIMEOUT = 3.0
# Product -> Tag slug mapping (future-proof for fox200 later)
# Product -> Tag slug mapping (kept unchanged, though not used now)
PRODUCT_TAG_SLUG = {
"fox100": "fox100-auto-upgrade-latest",
# "fox200": "fox200-auto-upgrade-latest",
@@ -156,7 +159,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
return None, 597
# NEW: RabbitMQ management API POST helper (basic auth; JSON in/out)
# RabbitMQ management API POST helper (basic auth; JSON in/out)
def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, timeout: float = RMQ_TIMEOUT):
body = json.dumps(payload_obj).encode("utf-8")
headers = {"Content-Type": "application/json"}
@@ -187,15 +190,15 @@ 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]]]:
def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any]]:
"""
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set)
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, upgrade_cmd)
- Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail).
- If device detail fetch fails, returns host/id with status/tags as None (treated as "no tag info").
- If device detail fetch fails, returns host/id with status/tags/upgrade_cmd as None.
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
return None, None, None, None, None
return None, None, None, None, None, None
base = NB_URL.rstrip("/")
h = {
@@ -208,12 +211,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
return 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
return None, None, None, None, None, None
rec = results[0]
aot = (rec.get("assigned_object_type") or "").strip()
@@ -224,29 +227,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
return 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
return None, aoid, 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
return None, aoid, 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
return None, aoid, None, None, None, None
# Step 3: Device detail (single fetch for BOTH status and tags)
# Step 3: Device detail (single fetch for status, tags, custom_fields.upgrade_cmd)
device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h)
if code3 != 200 or not device:
# treat as "no tag info" (no problem increment)
return host, aoid, dev_id, None, None
# treat as "no extra info"
return host, aoid, dev_id, None, None, None
status_val = ((device.get("status") or {}).get("value")) or None
tags = device.get("tags") or []
@@ -256,7 +259,10 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option
if isinstance(slug, str):
tag_slugs.add(slug)
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs
cf = device.get("custom_fields") or {}
upgrade_cmd = cf.get("upgrade_cmd")
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, upgrade_cmd
# =========================
@@ -330,30 +336,40 @@ async def main():
pass
host_suffix = ""
action_suffix = ""
bell_prefix = "" # ASCII BEL when we have a tag match
action_suffix = "" # kept; no longer used for tag action
bell_prefix = "" # ASCII BEL when upgrade_cmd present (3x)
netbox_time_ms = 0.0
desired_slug = PRODUCT_TAG_SLUG.get(product)
# NOTE: no tag check anymore; behavior depends on upgrade_cmd only
if product == "fox100":
nb_start = time.perf_counter()
try:
host, iface_id, dev_id, status_val, tag_slugs = nb_lookup_device_by_mac(mac=mac, log_status=log_status)
host, iface_id, dev_id, status_val, tag_slugs, upgrade_cmd = nb_lookup_device_by_mac(mac=mac, log_status=log_status)
if host:
host_suffix = f" host={host}"
if iface_id and not host:
host_suffix += f" iface_id={iface_id}"
# 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 desired_slug in tag_slugs:
bell_prefix = "\a" * 5
action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
# NEW: Publish immediate RMQ message like rmq-ikeja-pub3.sh (no delay)
# NEW LOGIC:
# - If host known:
# * upgrade_cmd empty/absent -> publish with task_name="ot-updater"
# * upgrade_cmd present -> publish with task_name="sot-updater-upgradecmd" and ring 3 BELs
if host:
try:
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
has_upgrade_cmd = False
try:
# consider non-empty string or any truthy value as "present"
if isinstance(upgrade_cmd, str):
has_upgrade_cmd = len(upgrade_cmd.strip()) > 0
else:
has_upgrade_cmd = bool(upgrade_cmd)
except Exception:
has_upgrade_cmd = False
task_name = "sot-updater-upgradecmd" if has_upgrade_cmd else "ot-updater"
rmq_body = {
"properties": {
"content_type": "application/json"
@@ -361,7 +377,7 @@ async def main():
"routing_key": RMQ_ROUTING_KEY,
"payload": json.dumps({
"inscope_device": host,
"task_name": "sot-updater-upgradecmd",
"task_name": task_name,
}),
"payload_encoding": "string",
}
@@ -369,7 +385,6 @@ async def main():
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:
routed = bool((resp or {}).get("routed", False))
@@ -377,6 +392,11 @@ async def main():
routed = False
if not routed:
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
# Bell behavior: ring 3x BEL only if upgrade_cmd present
if has_upgrade_cmd:
bell_prefix = "\a" * 3
except Exception as e:
await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
@@ -396,7 +416,7 @@ async def main():
line += f" subject={msg.subject}"
async with print_lock:
# Prepend BEL only when we had a tag match
# Prepend BEL only when we had upgrade_cmd present (3x)
sys.stdout.write(bell_prefix + line + "\n")
sys.stdout.flush()