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

1745

Browse Source
This commit is contained in:
pavel
2026-02-11 17:45:13 +02:00
parent 5c03bce9cc
commit 7004c0b9bb
1 changed files with 179 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
files/nats_registration_listener.py
View File

@@ -6,10 +6,11 @@ NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing
- Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting
- For fox100:
* If action_next is empty/absent -> print a simple stdout note and do nothing else
* If action_next is present -> PREPARE RabbitMQ /publish body and log:
"ok, here i will execute <body + decoded payload>"
(and publish to RabbitMQ)
* After successful publish (HTTP 200 + routed true): copy action_next -> action_last in NetBox (PATCH)
* If action_next present and action_last != action_next -> publish task_name=action_next
and on success set action_last=action_next and action_next_timestamp=now_epoch
* If action_next present and action_last == action_next -> publish only if
(now_epoch - action_next_timestamp) >= 600; if timestamp missing/invalid -> allow publish
and on success set action_last=action_next and action_next_timestamp=now_epoch
* If action_next present -> prepend 3x ASCII BEL to stdout line (kept behavior)
"""
@@ -214,15 +215,17 @@ 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]], Optional[Any]]:
def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any], Optional[Any], Optional[Any]
]:
"""
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next)
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next, action_last, action_next_timestamp)
- Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail).
- If device detail fetch fails, returns host/id with status/tags/action_next as None.
- If device detail fetch fails, returns host/id with status/tags/custom_fields as None.
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
return None, None, None, None, None, None
return None, None, None, None, None, None, None, None
base = NB_URL.rstrip("/")
h = {
@@ -235,12 +238,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, None
return None, None, 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, None
return None, None, None, None, None, None, None, None
rec = results[0]
aot = (rec.get("assigned_object_type") or "").strip()
@@ -251,29 +254,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, None
return None, None, 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, None
return None, aoid, None, None, 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, None
return None, aoid, None, None, 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, None
return None, aoid, None, None, None, None, None, None
# Step 3: Device detail (single fetch for status, tags, custom_fields.action_next)
# Step 3: Device detail (single fetch for status, tags, custom_fields.*)
device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h)
if code3 != 200 or not device:
# treat as "no extra info"
return host, aoid, dev_id, None, None, None
return host, aoid, dev_id, None, None, None, None, None
status_val = ((device.get("status") or {}).get("value")) or None
tags = device.get("tags") or []
@@ -285,8 +288,10 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Option
cf = device.get("custom_fields") or {}
action_next = cf.get("action_next")
action_last = cf.get("action_last")
action_next_timestamp = cf.get("action_next_timestamp")
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next, action_last, action_next_timestamp
# =========================
@@ -367,7 +372,9 @@ async def main():
if product == "fox100":
nb_start = time.perf_counter()
try:
host, iface_id, dev_id, status_val, tag_slugs, action_next = nb_lookup_device_by_mac(mac=mac, log_status=log_status)
host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp = nb_lookup_device_by_mac(
mac=mac, log_status=log_status
)
if host:
host_suffix = f" host={host}"
if iface_id and not host:
@@ -394,7 +401,44 @@ async def main():
async with print_lock:
print(f"[{ts()}] no action_next for host={host}", file=sys.stdout, flush=True)
else:
# Prepare RabbitMQ message with task_name taken from action_next (do not keep old task names)
# Compare action_last with action_next (strings)
action_last_str = None
try:
if isinstance(action_last, str):
action_last_str = action_last.strip()
elif action_last is None:
action_last_str = None
else:
action_last_str = str(action_last)
except Exception:
action_last_str = None
now_epoch = int(time.time())
# If action_last == action_next, apply cooldown based on action_next_timestamp (600s)
if action_last_str == action_next_str:
allow_repeat = True
try:
if action_next_timestamp is None:
allow_repeat = True
elif isinstance(action_next_timestamp, (int, float)):
allow_repeat = (now_epoch - int(action_next_timestamp)) >= 600
elif isinstance(action_next_timestamp, str):
allow_repeat = (now_epoch - int(action_next_timestamp.strip())) >= 600
else:
allow_repeat = True
except Exception:
allow_repeat = True
if not allow_repeat:
async with print_lock:
print(
f"[{ts()}] cooldown action_next for host={host} task={action_next_str}",
file=sys.stdout,
flush=True,
)
else:
# Publish task_name=action_next
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
payload_obj = {
@@ -413,7 +457,6 @@ async def main():
"payload_encoding": "string",
}
# Visual, easy-to-read log of what would be executed
await log_status(
f"[{ts()}] ok, here i will execute\n"
f" url: {rmq_url}\n"
@@ -422,7 +465,6 @@ async def main():
f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}"
)
# Publish
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}")
@@ -435,21 +477,85 @@ async def main():
if not routed:
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
else:
# Copy action_next -> action_last in NetBox (surgical)
# On success: set action_last and action_next_timestamp
try:
base = NB_URL.rstrip("/")
nb_headers = {
"Accept": "application/json",
"Authorization": f"Token {NB_TOKEN}",
}
patch_body = {"custom_fields": {"action_last": action_next_str}}
_, pcode = http_patch_json(f"{base}/api/dcim/devices/{dev_id}/", patch_body, headers=nb_headers, timeout=NB_TIMEOUT)
patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": now_epoch}}
_, pcode = http_patch_json(
f"{base}/api/dcim/devices/{dev_id}/",
patch_body,
headers=nb_headers,
timeout=NB_TIMEOUT,
)
if pcode != 200:
await log_status(f"[{ts()}] nb: action_last patch http={pcode} host={host} dev_id={dev_id}")
await log_status(f"[{ts()}] nb: action_last/timestamp patch http={pcode} host={host} dev_id={dev_id}")
except Exception as e:
await log_status(f"[{ts()}] nb: action_last patch error host={host!r} dev_id={dev_id!r} err={e!r}")
await log_status(f"[{ts()}] nb: action_last/timestamp patch error host={host!r} dev_id={dev_id!r} err={e!r}")
bell_prefix = "\a" * 3
else:
# action_last != action_next -> publish immediately
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
payload_obj = {
"inscope_device": host,
"task_name": action_next_str,
}
payload_raw = json.dumps(payload_obj, separators=(",", ":"), ensure_ascii=False)
rmq_body = {
"properties": {
"content_type": "application/json"
},
"routing_key": RMQ_ROUTING_KEY,
"payload": payload_raw,
"payload_encoding": "string",
}
await log_status(
f"[{ts()}] ok, here i will execute\n"
f" url: {rmq_url}\n"
f" routing_key: {RMQ_ROUTING_KEY}\n"
f" payload_raw: {payload_raw}\n"
f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}"
)
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:
routed = False
try:
routed = bool((resp or {}).get("routed", False))
except Exception:
routed = False
if not routed:
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
else:
now_epoch = int(time.time())
try:
base = NB_URL.rstrip("/")
nb_headers = {
"Accept": "application/json",
"Authorization": f"Token {NB_TOKEN}",
}
patch_body = {"custom_fields": {"action_last": action_next_str, "action_next_timestamp": now_epoch}}
_, pcode = http_patch_json(
f"{base}/api/dcim/devices/{dev_id}/",
patch_body,
headers=nb_headers,
timeout=NB_TIMEOUT,
)
if pcode != 200:
await log_status(f"[{ts()}] nb: action_last/timestamp patch http={pcode} host={host} dev_id={dev_id}")
except Exception as e:
await log_status(f"[{ts()}] nb: action_last/timestamp patch error host={host!r} dev_id={dev_id!r} err={e!r}")
# Bell behavior: ring 3x BEL when action_next present
bell_prefix = "\a" * 3
except Exception as e: