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

0851

Browse Source
This commit is contained in:
pavel
2026-02-12 08:51:54 +02:00
parent 07c1a9bb22
commit 99bc927ea2
1 changed files with 38 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
files/nats_registration_listener.py
View File

@@ -216,16 +216,16 @@ async def nb_problem(log_status, msg: 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], Optional[Any], Optional[Any]
Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any], Optional[Any], Optional[Any], Optional[Any]
]:
"""
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next, action_last, action_next_timestamp)
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next, action_last, action_next_timestamp, action_state)
- Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail).
- 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, None, None
return None, None, None, None, None, None, None, None, None
base = NB_URL.rstrip("/")
h = {
@@ -238,12 +238,12 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
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, None, None
return None, 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, None, None
return None, None, None, None, None, None, None, None, None
rec = results[0]
aot = (rec.get("assigned_object_type") or "").strip()
@@ -254,29 +254,29 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
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, None, None
return None, 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, None, None
return None, aoid, None, 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, None, None
return None, aoid, None, 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, None, None
return None, aoid, None, None, None, None, None, None, None
# 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, None, None
return host, aoid, dev_id, None, None, None, None, None, None
status_val = ((device.get("status") or {}).get("value")) or None
tags = device.get("tags") or []
@@ -290,8 +290,9 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
action_next = cf.get("action_next")
action_last = cf.get("action_last")
action_next_timestamp = cf.get("action_next_timestamp")
action_state = cf.get("action_state")
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next, action_last, action_next_timestamp
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs, action_next, action_last, action_next_timestamp, action_state
# =========================
@@ -372,7 +373,7 @@ async def main():
if product == "fox100":
nb_start = time.perf_counter()
try:
host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp = nb_lookup_device_by_mac(
host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp, action_state = nb_lookup_device_by_mac(
mac=mac, log_status=log_status
)
if host:
@@ -381,6 +382,27 @@ async def main():
host_suffix += f" iface_id={iface_id}"
if host:
# Gate on action_state: allow only "" or "ready"
action_state_str = ""
try:
if action_state is None:
action_state_str = ""
elif isinstance(action_state, str):
action_state_str = action_state.strip()
else:
action_state_str = str(action_state).strip()
except Exception:
action_state_str = ""
if action_state_str not in ("", "ready"):
async with print_lock:
print(
f"[{ts()}] device is not ready because of action_state host={host} action_state={action_state_str}",
file=sys.stdout,
flush=True,
)
return
# Determine if action_next is present (non-empty string, or any truthy value)
has_action_next = False
action_next_str = None
@@ -477,14 +499,14 @@ async def main():
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
# On success: set action_last and action_next_timestamp and action_state
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": str(now_epoch)}}
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,
@@ -544,7 +566,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)}}
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,
@@ -608,4 +630,4 @@ if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
pass