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

06:25

Browse Source
This commit is contained in:
pavel
2025-10-31 06:25:53 +02:00
parent 62ac98ba21
commit 5db16e7eb7
1 changed files with 87 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
files/nats_registration_listener.py
View File

@@ -1,13 +1,15 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (with NetBox hostname lookup for fox100 + timing + problem counter)
----------------------------------------------------------------------------------------------
NATS Registration Listener (fox100 + NetBox hostname/tag lookup + timing + problem counter)
-------------------------------------------------------------------------------------------
Adds:
- 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 from message arrival till final output (total_ms)
- Increments problem counter for each NetBox anomaly or lookup issue
- Adds iface_id=... to stderr diagnostics
- Measures total latency (total_ms)
- Increments problem counter for NetBox anomalies, includes iface_id where applicable
"""
import argparse
@@ -21,7 +23,7 @@ import hashlib
import time
from time import monotonic
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from typing import Optional, Dict, Any, Tuple
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
@@ -36,7 +38,13 @@ NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL
NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided
NB_TIMEOUT = 3.0 # seconds per HTTP GET
# Global counter for any NetBox-related errors/anomalies
# 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
}
# Global counter for any NetBox-related problems
NB_PROBLEM_COUNTER = 0
NB_PROBLEM_LOCK = asyncio.Lock()
@@ -148,13 +156,15 @@ async def nb_problem(log_status, msg: str):
await log_status(msg)
def nb_get_hostname_by_mac(mac: str, log_status) -> (Optional[str], Optional[int]):
def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str]]:
"""
Return (hostname, iface_id or None)
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).
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
return None, None
return None, None, None, None
base = NB_URL.rstrip("/")
h = {
@@ -167,39 +177,75 @@ def nb_get_hostname_by_mac(mac: str, log_status) -> (Optional[str], Optional[int
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
return 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
if len(results) > 1:
# no iface_id yet; fetch it below
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: multiple mac records mac={mac_norm}"))
return 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
return 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
return None, aoid, None, None
# Step 2: Interface → Device
# Step 2: Interface -> Device (shallow device obj)
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
return None, aoid, None, None
dev = iface.get("device") or {}
host = dev.get("name") or dev.get("display")
if not host:
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
return None, aoid, None, None
return host, aoid
# Step 3: Device detail for status/tags
# NOTE: As requested, failures here are treated as "no tag info" (no problem increment).
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
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 []
for t in tags:
slug = t.get("slug")
if isinstance(slug, str) and slug == desired_slug:
return True
return False
# =========================
@@ -251,6 +297,7 @@ async def main():
t_start = monotonic()
payload = msg.data
# strict payload dedupe
digest = hashlib.blake2b(payload, digest_size=16).digest()
nowm = monotonic()
exp = recent_payloads.get(digest)
@@ -272,16 +319,32 @@ async def main():
pass
host_suffix = ""
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 = nb_get_hostname_by_mac(mac=mac, log_status=log_status)
host, iface_id, dev_id, status_val = 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
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
@@ -290,7 +353,7 @@ async def main():
async with NB_PROBLEM_LOCK:
nb_problems_snapshot = NB_PROBLEM_COUNTER
line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}"
line = f"nb_problems={nb_problems_snapshot} [{ts()}] product={product} mac={mac} fw={fw}{host_suffix}{action_suffix}"
if product == "fox100":
line += f" netbox_ms={netbox_time_ms:.1f} total_ms={total_ms:.1f}"
if args.include_subject: