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

feat: startign with netbox 06:57

Browse Source
This commit is contained in:
pavel
2025-10-30 06:57:08 +02:00
parent e4e8fb19a1
commit 8eed0e8cc5
1 changed files with 147 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
files/nats_registration_listener.py
View File

@@ -1,17 +1,22 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (compact one-line output + strict read-only dedupe)
------------------------------------------------------------------------------
Print exactly one line per message:
[TS] product=<productName> mac=<mac> fw=<active>
NATS Registration Listener (with optional NetBox hostname lookup for fox100)
----------------------------------------------------------------------------
- Listener-only (does NOT publish/reply).
- Writes are serialized to avoid interleaved output.
- Strict dedupe by exact payload bytes within a short TTL (env DEDUP_TTL, default 2.0s).
- Falls back gracefully if fields/structure are missing.
Adds: For product=fox100, lookup device hostname in NetBox by MAC and append
'host=<device.name>' to the one-line output if found.
Config (env or CLI):
NB_URL (e.g., http://netbox.gt-tiso.ikeja.co.za)
NB_TOKEN (API token)
NB_TIMEOUT (seconds, default 3.0)
Behavior:
- Non-fox100: unchanged.
- fox100: best-effort NetBox lookup inline (sequential), stderr on edge cases.
Install deps:
pip install --upgrade nats-py
pip install --upgrade nats-py aiohttp
"""
import argparse
@@ -27,7 +32,10 @@ from datetime import datetime, timezone
from typing import Optional, Dict, Any
import nats
import aiohttp
from aiohttp import ClientTimeout
# ----------------- existing helpers (unchanged) -----------------
def parse_args():
p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.")
@@ -59,24 +67,25 @@ def parse_args():
# Misc
p.add_argument("--include-subject", action="store_true", help="Append subject=... to each output line.")
# --- NetBox cfg ---
p.add_argument("--nb-url", default=os.environ.get("NB_URL"), help="NetBox base URL (env NB_URL).")
p.add_argument("--nb-token", default=os.environ.get("NB_TOKEN"), help="NetBox API token (env NB_TOKEN).")
p.add_argument("--nb-timeout", type=float, default=float(os.environ.get("NB_TIMEOUT", "3.0")), help="NetBox HTTP timeout seconds (default 3.0).")
return p.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
@@ -85,7 +94,6 @@ def ts() -> str:
def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]:
# Prefer eth0; otherwise first key deterministically (sorted)
if "eth0" in eths and isinstance(eths["eth0"], dict):
return eths["eth0"]
for name in sorted(eths.keys()):
@@ -95,30 +103,17 @@ 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:
# Some producers might not wrap with "data"
d = root
product = d.get("productName") or "-"
fw_active = "-"
try:
fw_active = d.get("firmwareVersion", {}).get("active") or "-"
except Exception:
pass
mac = "-"
try:
eths = d.get("ethernetInterfaces", {})
@@ -128,26 +123,108 @@ def extract_fields(obj: Dict[str, Any]):
mac = chosen["macAddress"]
except Exception:
pass
return product, mac, fw_active
# ----------------- new: NetBox helpers -----------------
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 dashes with colons
s = s.replace('-', ':')
hex_only = ''.join(ch for ch in s if ch in '0123456789abcdef')
if len(hex_only) == 12:
# reconstruct as colon-separated
s = ':'.join(hex_only[i:i+2] for i in range(0, 12, 2))
return s
# already colonized?
parts = s.split(':')
if len(parts) == 6 and all(len(p) == 2 and all(c in '0123456789abcdef' for c in p) for p in parts):
return s
return None
async def nb_get_hostname_by_mac(session: aiohttp.ClientSession, base_url: str, token: str, mac: str, timeout_s: float, log_status):
"""
Best-effort:
1) GET /api/dcim/mac-addresses/?mac_address=<mac>
2) If one result and assigned to interface -> GET /api/dcim/interfaces/{id}/ -> device.name
Returns (hostname or None).
Emits concise diagnostics to stderr via log_status on edge cases.
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
await log_status(f"[{ts()}] nb: invalid mac format mac={mac!r}")
return None
# 1) find MAC record(s)
url_mac = f"{base_url.rstrip('/')}/api/dcim/mac-addresses/"
try:
async with session.get(url_mac, params={"mac_address": mac_norm, "limit": 2}, headers={"Authorization": f"Token {token}"}, timeout=timeout_s) as r:
if r.status != 200:
await log_status(f"[{ts()}] nb: mac query http={r.status} mac={mac_norm}")
return None
data = await r.json()
except Exception as e:
await log_status(f"[{ts()}] nb: mac query error mac={mac_norm} err={e!r}")
return None
results = (data or {}).get("results") or []
if not results:
await log_status(f"[{ts()}] nb: mac not found mac={mac_norm}")
return None
if len(results) > 1:
await 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:
await log_status(f"[{ts()}] nb: mac unassigned mac={mac_norm}")
return None
if aot != "dcim.interface":
await log_status(f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm}")
return None
# 2) fetch interface -> device
url_iface = f"{base_url.rstrip('/')}/api/dcim/interfaces/{aoid}/"
try:
async with session.get(url_iface, headers={"Authorization": f"Token {token}"}, timeout=timeout_s) as r2:
if r2.status != 200:
await log_status(f"[{ts()}] nb: iface fetch http={r2.status} iface_id={aoid}")
return None
iface = await r2.json()
except Exception as e:
await log_status(f"[{ts()}] nb: iface fetch error iface_id={aoid} err={e!r}")
return None
dev = iface.get("device") or {}
host = dev.get("name") or dev.get("display")
if not host:
await log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}")
return None
return host
# ----------------- main -----------------
async def main():
args = parse_args()
ssl_ctx = make_ssl_context(args)
# Serialize stdout writes to avoid interleaving
print_lock = asyncio.Lock()
async def log_status(s: str):
async with print_lock:
print(s, file=sys.stderr, flush=True)
# --- Strict read-only dedupe config (env) ---
# Drop exact duplicate payloads seen within this TTL window.
# dedupe
DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0"))
recent_payloads: Dict[bytes, float] = {} # digest -> expires_at (monotonic seconds)
recent_payloads: Dict[bytes, float] = {}
reconnect_time_wait = 2
max_reconnect_attempts = -1
@@ -180,29 +257,33 @@ async def main():
tls=ssl_ctx,
)
# --- aiohttp session for NetBox ---
nb_enabled = bool(args.nb_url and args.nb_token)
if nb_enabled:
timeout = ClientTimeout(total=None) # per-call timeout set explicitly
http = aiohttp.TCPConnector(limit=32)
session = aiohttp.ClientSession(timeout=timeout, connector=http)
else:
session = None
async def message_handler(msg: nats.aio.msg.Msg):
now_s = ts()
payload = msg.data
# --- STRICT DEDUPE by exact payload bytes (read-only) ---
# Use a stable, compact digest to key the recent map.
# dedupe by payload
digest = hashlib.blake2b(payload, digest_size=16).digest()
nowm = monotonic()
exp = recent_payloads.get(digest)
if exp and exp > nowm:
return # drop exact duplicate seen very recently
return
recent_payloads[digest] = nowm + DEDUPE_TTL
# Optional light cleanup to keep the dict bounded
if len(recent_payloads) > 4096:
# remove expired entries
cutoff = nowm
recent_payloads_keys = list(recent_payloads.keys())
for k in recent_payloads_keys:
for k in list(recent_payloads.keys()):
if recent_payloads.get(k, 0) <= cutoff:
recent_payloads.pop(k, None)
# Parse and print one line
# parse payload
product = mac = fw = "-"
try:
text = payload.decode("utf-8", errors="replace")
@@ -211,25 +292,42 @@ async def main():
except Exception:
pass
line = f"[{now_s}] product={product} mac={mac} fw={fw}"
# optional NetBox lookup for fox100
host_suffix = ""
if nb_enabled and product == "fox100":
try:
host = await nb_get_hostname_by_mac(
session=session,
base_url=args.nb_url,
token=args.nb_token,
mac=mac,
timeout_s=args.nb_timeout,
log_status=log_status,
)
if host:
host_suffix = f" host={host}"
except Exception as e:
await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
# print one line
line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}"
if args.include_subject:
line += f" subject={msg.subject}"
async with print_lock:
sys.stdout.write(line + "\n")
sys.stdout.flush()
# subscribe
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"
f"[{ts()}] Listening on '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} | mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled nb={'on' if nb_enabled else 'off'}"
)
# Graceful shutdown
# graceful shutdown
stop_event = asyncio.Event()
def handle_signal(*_):
@@ -248,7 +346,8 @@ async def main():
await nc.drain()
finally:
await nc.close()
if session:
await session.close()
if __name__ == "__main__":
try: