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

feat: startign with netbox 07:10

Browse Source
This commit is contained in:
pavel
2025-10-30 07:10:27 +02:00
parent beb31c9429
commit 62ac98ba21
1 changed files with 93 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
files/nats_registration_listener.py
View File

@@ -1,32 +1,13 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (with NetBox hostname lookup for fox100, no extra deps)
----------------------------------------------------------------------------------
Behavior:
- Prints exactly one line per message:
[TS] product=<productName> mac=<mac> fw=<active> [host=<device.name> for fox100]
- Listener-only (no publish/reply).
- Payload-byte strict dedupe within a TTL window.
- Graceful shutdown + reconnect logging.
- For product=fox100: best-effort NetBox lookup by MAC -> interface -> device.name.
(Inline, sequential "as-is" call; no concurrency / worker pool.)
NetBox (hardcoded as requested):
- Edit NB_TOKEN below.
- Uses stdlib urllib (no 'requests' / 'aiohttp').
CLI (same spirit as before; your entrypoint passes these):
--servers nats://host:4222 [one or more]
--subject v1.registration
--queue <group> (optional)
--name registration-listener
--tls-ca /path/to/ca.pem
--tls-cert/--tls-key/--insecure
--include-subject
Install deps in image:
pip install --no-cache-dir --upgrade nats-py
NATS Registration Listener (with NetBox hostname lookup for fox100 + timing + problem counter)
----------------------------------------------------------------------------------------------
Adds:
- nb_problems=<count> prefix on every stdout line
- 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
"""
import argparse
@@ -37,6 +18,7 @@ import signal
import ssl
import sys
import hashlib
import time
from time import monotonic
from datetime import datetime, timezone
from typing import Optional, Dict, Any
@@ -51,43 +33,32 @@ import nats
# NetBox hardcoded config
# =========================
NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL
NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # <-- paste the real token here
NB_TIMEOUT = 3.0 # seconds per HTTP GET
NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided
NB_TIMEOUT = 3.0 # seconds per HTTP GET
# Global counter for any NetBox-related errors/anomalies
NB_PROBLEM_COUNTER = 0
NB_PROBLEM_LOCK = asyncio.Lock()
# =========================
# Arg parsing (mirrors original)
# Arg parsing
# =========================
def parse_args():
p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.")
p.add_argument(
"--servers",
nargs="+",
help="One or more NATS server URLs (e.g., nats://127.0.0.1:4222).",
default=["nats://127.0.0.1:4222"],
)
p.add_argument(
"--subject",
help="NATS subject to subscribe to.",
default="registration",
)
p.add_argument("--queue", help="Optional queue group name.", default=None)
p.add_argument("--name", help="Client connection name.", default="registration-listener")
# Auth
p.add_argument("--creds", help="Path to .creds file (JWT + NKey).")
p.add_argument("--user", help="Username.")
p.add_argument("--password", help="Password.")
p.add_argument("--token", help="Auth token.")
# TLS
p.add_argument("--tls-ca", help="Path to CA certificate for TLS.")
p.add_argument("--tls-cert", help="Path to client certificate for TLS.")
p.add_argument("--tls-key", help="Path to client key for TLS.")
p.add_argument("--insecure", action="store_true", help="Disable TLS hostname verification.")
# Misc
p.add_argument("--include-subject", action="store_true", help="Append subject=... to each output line.")
p.add_argument("--servers", nargs="+", default=["nats://127.0.0.1:4222"])
p.add_argument("--subject", default="registration")
p.add_argument("--queue", default=None)
p.add_argument("--name", default="registration-listener")
p.add_argument("--creds")
p.add_argument("--user")
p.add_argument("--password")
p.add_argument("--token")
p.add_argument("--tls-ca")
p.add_argument("--tls-cert")
p.add_argument("--tls-key")
p.add_argument("--insecure", action="store_true")
p.add_argument("--include-subject", action="store_true")
return p.parse_args()
@@ -97,23 +68,18 @@ def parse_args():
def make_ssl_context(args) -> Optional[ssl.SSLContext]:
if not any([args.tls_ca, args.tls_cert, args.tls_key]) and not any(url.startswith("tls://") for url in args.servers):
return None
ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
if args.tls_ca:
ctx.load_verify_locations(args.tls_ca)
if args.tls_cert and args.tls_key:
ctx.load_cert_chain(certfile=args.tls_cert, keyfile=args.tls_key)
if args.insecure:
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
return ctx
def ts() -> str:
# local time with offset like your previous output
return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z")
@@ -127,50 +93,23 @@ def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]:
def extract_fields(obj: Dict[str, Any]):
"""
Expected structure (from examples):
{ "data": {
"productName": "...",
"ethernetInterfaces": { "eth0": { "macAddress": "..." } },
"firmwareVersion": { "active": "..." }
},
"metadata": { "type": "REGISTRATION_REQUEST" }
}
"""
root = obj
if isinstance(root.get("data"), dict):
d = root["data"]
else:
d = root
d = root["data"] if isinstance(root.get("data"), dict) else root
product = d.get("productName") or "-"
fw_active = "-"
try:
fw_active = d.get("firmwareVersion", {}).get("active") or "-"
except Exception:
pass
fw_active = d.get("firmwareVersion", {}).get("active") or "-"
mac = "-"
try:
eths = d.get("ethernetInterfaces", {})
if isinstance(eths, dict):
chosen = pick_first_interface(eths)
if chosen and isinstance(chosen.get("macAddress"), str):
mac = chosen["macAddress"]
except Exception:
pass
eths = d.get("ethernetInterfaces", {})
if isinstance(eths, dict):
chosen = pick_first_interface(eths)
if chosen and isinstance(chosen.get("macAddress"), str):
mac = chosen["macAddress"]
return product, mac, fw_active
# =========================
# NetBox lookup (stdlib urllib)
# NetBox lookup (urllib)
# =========================
def normalize_mac(mac: str) -> Optional[str]:
"""
Return lowercase colon-separated MAC or None if invalid-ish.
Accepts colon, dash, or bare hex and tries to normalize.
"""
if not mac or not isinstance(mac, str):
return None
s = mac.strip().lower().replace("-", ":")
@@ -192,10 +131,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
if resp.status != 200:
return None, resp.status
data = resp.read()
try:
return json.loads(data.decode("utf-8", errors="replace")), 200
except Exception:
return None, 200
return json.loads(data.decode("utf-8", errors="replace")), 200
except HTTPError as e:
return None, getattr(e, "code", 599)
except URLError:
@@ -204,18 +140,21 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
return None, 597
def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]:
async def nb_problem(log_status, msg: str):
"""Increment counter and log a problem line."""
global NB_PROBLEM_COUNTER
async with NB_PROBLEM_LOCK:
NB_PROBLEM_COUNTER += 1
await log_status(msg)
def nb_get_hostname_by_mac(mac: str, log_status) -> (Optional[str], Optional[int]):
"""
Steps:
1) GET /api/dcim/mac-addresses/?mac_address=<mac>
2) If assigned to dcim.interface -> GET /api/dcim/interfaces/{id}/
3) Return device.name
Emits concise diagnostics to stderr via log_status on edge cases.
Return (hostname, iface_id or None)
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
# invalid or missing mac; keep stdout clean
return None
return None, None
base = NB_URL.rstrip("/")
h = {
@@ -224,42 +163,43 @@ def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]:
"Authorization": f"Token {NB_TOKEN}",
}
# 1) find MAC record(s)
# Step 1: MAC lookup
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:
# network/HTTP issue or parse fail
asyncio.create_task(log_status(f"[{ts()}] nb: mac query http={code} mac={mac_norm}"))
return None
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac query http={code} mac={mac_norm}"))
return None, None
results = (data or {}).get("results") or []
if not results:
asyncio.create_task(log_status(f"[{ts()}] nb: mac not found mac={mac_norm}"))
return None
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac not found mac={mac_norm}"))
return None, None
if len(results) > 1:
asyncio.create_task(log_status(f"[{ts()}] nb: multiple mac records mac={mac_norm}"))
# no iface_id yet; fetch it below
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: multiple mac records mac={mac_norm}"))
rec = results[0]
aot = (rec.get("assigned_object_type") or "").strip()
aoid = rec.get("assigned_object_id")
if not aot or aoid is None:
asyncio.create_task(log_status(f"[{ts()}] nb: mac unassigned mac={mac_norm}"))
return None
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac unassigned mac={mac_norm}"))
return None, None
if aot != "dcim.interface":
asyncio.create_task(log_status(f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm}"))
return None
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm} iface_id={aoid}"))
return None, aoid
# 2) fetch interface -> device
# Step 2: Interface → Device
iface, code2 = http_get_json(f"{base}/api/dcim/interfaces/{aoid}/", headers=h)
if code2 != 200 or not iface:
asyncio.create_task(log_status(f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}"))
return None
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}"))
return None, aoid
dev = iface.get("device") or {}
host = dev.get("name") or dev.get("display")
if not host:
asyncio.create_task(log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}"))
return None
return host
asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: iface has no device iface_id={aoid}"))
return None, aoid
return host, aoid
# =========================
@@ -275,13 +215,9 @@ async def main():
async with print_lock:
print(s, file=sys.stderr, flush=True)
# --- Strict read-only dedupe (payload bytes) ---
DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) # you can tune via env if you want; otherwise 2.0s
DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0"))
recent_payloads: Dict[bytes, float] = {}
reconnect_time_wait = 2
max_reconnect_attempts = -1
async def disconnected_cb():
await log_status(f"[{ts()}] Disconnected from NATS.")
@@ -298,8 +234,8 @@ async def main():
servers=args.servers,
name=args.name,
allow_reconnect=True,
reconnect_time_wait=reconnect_time_wait,
max_reconnect_attempts=max_reconnect_attempts,
reconnect_time_wait=2,
max_reconnect_attempts=-1,
disconnected_cb=disconnected_cb,
reconnected_cb=reconnected_cb,
error_cb=error_cb,
@@ -312,10 +248,9 @@ async def main():
)
async def message_handler(msg: nats.aio.msg.Msg):
now_s = ts()
t_start = monotonic()
payload = msg.data
# --- dedupe by exact payload bytes ---
digest = hashlib.blake2b(payload, digest_size=16).digest()
nowm = monotonic()
exp = recent_payloads.get(digest)
@@ -325,10 +260,9 @@ async def main():
if len(recent_payloads) > 4096:
cutoff = nowm
for k in list(recent_payloads.keys()):
if recent_payloads.get(k, 0) <= cutoff:
if recent_payloads[k] <= cutoff:
recent_payloads.pop(k, None)
# parse
product = mac = fw = "-"
try:
text = payload.decode("utf-8", errors="replace")
@@ -337,39 +271,43 @@ async def main():
except Exception:
pass
# optional NetBox lookup for fox100 (inline, sequential)
host_suffix = ""
netbox_time_ms = 0.0
if product == "fox100":
nb_start = time.perf_counter()
try:
host = nb_get_hostname_by_mac(mac=mac, log_status=log_status)
host, iface_id = nb_get_hostname_by_mac(mac=mac, log_status=log_status)
if host:
host_suffix = f" host={host}"
else:
# minimal diag already sent to stderr by helper
pass
if iface_id and not host:
host_suffix += f" iface_id={iface_id}"
except Exception as e:
await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
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
# print one compact line
line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}"
total_ms = (monotonic() - t_start) * 1000
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}"
if product == "fox100":
line += f" netbox_ms={netbox_time_ms:.1f} total_ms={total_ms:.1f}"
if args.include_subject:
line += f" subject={msg.subject}"
async with print_lock:
sys.stdout.write(line + "\n")
sys.stdout.flush()
# subscribe
async with print_lock:
print(line, flush=True)
if args.queue:
await nc.subscribe(args.subject, queue=args.queue, cb=message_handler)
else:
await nc.subscribe(args.subject, cb=message_handler)
await log_status(
f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} "
f"| mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled nb=on"
f"[{ts()}] Listening on subject '{args.subject}' (queue={args.queue or '-'}) via {args.servers} | nb=on"
)
# graceful shutdown
stop_event = asyncio.Event()
def handle_signal(*_):
@@ -384,10 +322,8 @@ async def main():
signal.signal(s, lambda *_: handle_signal())
await stop_event.wait()
try:
await nc.drain()
finally:
await nc.close()
await nc.drain()
await nc.close()
if __name__ == "__main__":