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

06:33

Browse Source
This commit is contained in:
pavel
2025-10-31 06:33:09 +02:00
parent 5db16e7eb7
commit 9498ab3016
1 changed files with 31 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
files/nats_registration_listener.py
View File

@@ -2,13 +2,15 @@
"""
NATS Registration Listener (fox100 + NetBox hostname/tag lookup + timing + problem counter)
-------------------------------------------------------------------------------------------
Adds:
Removes duplicate device GET:
- Single /devices/{id}/ fetch provides both status and tags.
Adds/keeps:
- nb_problems=<count> prefix on every stdout line
- For product=fox100: MAC -> interface -> device lookup in NetBox (urllib only)
- Device tag check for slug 'fox100-auto-upgrade-latest' when status=active
- If tag present & active: append "action=ok, i'm ready to schedule this device <hostname> upgrade"
- Measures per-fox100 NetBox lookup duration (netbox_ms)
- Measures total latency (total_ms)
- Measures per-fox100 NetBox lookup duration (netbox_ms) and total latency (total_ms)
- Increments problem counter for NetBox anomalies, includes iface_id where applicable
"""
@@ -23,7 +25,7 @@ import hashlib
import time
from time import monotonic
from datetime import datetime, timezone
from typing import Optional, Dict, Any, Tuple
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
@@ -41,7 +43,7 @@ NB_TIMEOUT = 3.0 # seconds per HTTP GET
# Product -> Tag slug mapping (future-proof for fox200 later)
PRODUCT_TAG_SLUG = {
"fox100": "fox100-auto-upgrade-latest",
# "fox200": "fox200-auto-upgrade-latest", # placeholder for later
# "fox200": "fox200-auto-upgrade-latest",
}
# Global counter for any NetBox-related problems
@@ -156,15 +158,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]]:
def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]]]:
"""
Resolve MAC -> (hostname, iface_id, device_id, device_status_value)
- Logs problems (async) for anomalies, but per Pavel's ask we treat device-fetch failures as "no tag info"
(i.e., we WON'T count those as problems unless it's a clear anomaly like mac not found / unassigned / wrong type).
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set)
- 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").
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
return None, None, None, None
return None, None, None, None, None
base = NB_URL.rstrip("/")
h = {
@@ -177,75 +179,55 @@ 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
return 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
return None, None, None, None, None
rec = results[0]
aot = (rec.get("assigned_object_type") or "").strip()
aoid = rec.get("assigned_object_id")
# If multiple results, report with iface_id if we can infer one from the first record
if len(results) > 1:
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: multiple mac records mac={mac_norm} iface_id={aoid if aoid is not None else '-'}"))
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
return None, None, None, None, None
if aot != "dcim.interface":
# We can still print the aoid for context
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
return None, aoid, None, None, None
# Step 2: Interface -> Device (shallow device obj)
# 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
return None, aoid, 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
return None, aoid, None, None, None
# Step 3: Device detail for status/tags
# NOTE: As requested, failures here are treated as "no tag info" (no problem increment).
# Step 3: Device detail (single fetch now provides BOTH status and tags)
device, code3 = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h)
if code3 != 200 or not device:
# Treat as no tag data; return host/id so upstream can still print hostname
return host, aoid, dev_id, None
# treat as "no tag info" (no problem increment per Pavel's guidance)
return host, aoid, dev_id, None, None
status_val = ((device.get("status") or {}).get("value")) or None
return host, aoid, dev_id, status_val if isinstance(status_val, str) else None
def nb_device_has_tag(dev_id: int, desired_slug: str) -> Optional[bool]:
"""
Return True/False if we can reliably determine presence of the tag by slug on the device.
Return None if we couldn't fetch/parse (treat as 'not present' by caller).
"""
if dev_id is None:
return None
base = NB_URL.rstrip("/")
h = {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": f"Token {NB_TOKEN}",
}
device, code = http_get_json(f"{base}/api/dcim/devices/{dev_id}/", headers=h)
if code != 200 or not device:
return None
tags = device.get("tags") or []
tag_slugs = set()
for t in tags:
slug = t.get("slug")
if isinstance(slug, str) and slug == desired_slug:
return True
return False
if isinstance(slug, str):
tag_slugs.add(slug)
return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs
# =========================
@@ -322,29 +304,21 @@ async def main():
action_suffix = ""
netbox_time_ms = 0.0
# fox100 logic (future: can generalize by product -> tag)
desired_slug = PRODUCT_TAG_SLUG.get(product)
if product == "fox100":
nb_start = time.perf_counter()
host = None
dev_id = None
try:
host, iface_id, dev_id, status_val = nb_lookup_device_by_mac(mac=mac, log_status=log_status)
host, iface_id, dev_id, status_val, tag_slugs = 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}"
# Only attempt tag check if we have a device id AND status=active AND a desired slug
if dev_id is not None and status_val == "active" and isinstance(desired_slug, str):
has_tag = nb_device_has_tag(dev_id, desired_slug)
if has_tag is True:
# CAP: print readiness to schedule upgrade (no side effects yet)
# Keep it on the same stdout line
# Only attempt the tag-based action if we have device info and it's active
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:
action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
# If has_tag is False/None, just continue silently
except Exception as e:
# Unexpected NetBox error: count as a problem and continue
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