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

feat: startign with netbox 07:02

Browse Source
This commit is contained in:
pavel
2025-10-30 07:02:31 +02:00
parent 8eed0e8cc5
commit acdebf568e
1 changed files with 133 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

225
files/nats_registration_listener.py
View File

@@ -1,22 +1,32 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (with optional NetBox hostname lookup for fox100)
----------------------------------------------------------------------------
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)
NATS Registration Listener (with NetBox hostname lookup for fox100, no extra deps)
----------------------------------------------------------------------------------
Behavior:
- Non-fox100: unchanged.
- fox100: best-effort NetBox lookup inline (sequential), stderr on edge cases.
- 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.)
Install deps:
pip install --upgrade nats-py aiohttp
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
"""
import argparse
@@ -30,27 +40,38 @@ import hashlib
from time import monotonic
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import nats
import aiohttp
from aiohttp import ClientTimeout
# ----------------- existing helpers (unchanged) -----------------
# =========================
# NetBox hardcoded config
# =========================
NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL
NB_TOKEN = "CHANGE_ME_NETBOX_TOKEN" # <-- paste the real token here
NB_TIMEOUT = 3.0 # seconds per HTTP GET
# =========================
# Arg parsing (mirrors original)
# =========================
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=[os.environ.get("NATS_URL", "nats://127.0.0.1:4222")],
default=["nats://127.0.0.1:4222"],
)
p.add_argument(
"--subject",
help="NATS subject to subscribe to.",
default=os.environ.get("NATS_SUBJECT", "registration"),
default="registration",
)
p.add_argument("--queue", help="Optional queue group name.", default=os.environ.get("NATS_QUEUE"))
p.add_argument("--queue", help="Optional queue group name.", default=None)
p.add_argument("--name", help="Client connection name.", default="registration-listener")
# Auth
@@ -67,29 +88,32 @@ 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()
# =========================
# Helpers
# =========================
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")
@@ -103,17 +127,29 @@ 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
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", {})
@@ -123,10 +159,13 @@ def extract_fields(obj: Dict[str, Any]):
mac = chosen["macAddress"]
except Exception:
pass
return product, mac, fw_active
# ----------------- new: NetBox helpers -----------------
# =========================
# NetBox lookup (stdlib urllib)
# =========================
def normalize_mac(mac: str) -> Optional[str]:
"""
Return lowercase colon-separated MAC or None if invalid-ish.
@@ -134,84 +173,98 @@ def normalize_mac(mac: str) -> Optional[str]:
"""
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')
s = mac.strip().lower().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 ":".join(hex_only[i:i+2] for i in range(0, 12, 2))
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):
def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, timeout: float = NB_TIMEOUT):
if params:
url = f"{url}?{urlencode(params)}"
req = Request(url, headers=headers or {}, method="GET")
try:
with urlopen(req, timeout=timeout) as resp:
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
except HTTPError as e:
return None, getattr(e, "code", 599)
except URLError:
return None, 598
except Exception:
return None, 597
def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]:
"""
Best-effort:
Steps:
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).
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.
"""
mac_norm = normalize_mac(mac)
if not mac_norm:
await log_status(f"[{ts()}] nb: invalid mac format mac={mac!r}")
# invalid or missing mac; keep stdout clean
return None
base = NB_URL.rstrip("/")
h = {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": f"Token {NB_TOKEN}",
}
# 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}")
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
results = (data or {}).get("results") or []
if not results:
await log_status(f"[{ts()}] nb: mac not found mac={mac_norm}")
asyncio.create_task(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}")
asyncio.create_task(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}")
asyncio.create_task(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}")
asyncio.create_task(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}")
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
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}")
asyncio.create_task(log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}"))
return None
return host
# ----------------- main -----------------
# =========================
# Main
# =========================
async def main():
args = parse_args()
ssl_ctx = make_ssl_context(args)
@@ -222,9 +275,10 @@ async def main():
async with print_lock:
print(s, file=sys.stderr, flush=True)
# dedupe
DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0"))
# --- 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
recent_payloads: Dict[bytes, float] = {}
reconnect_time_wait = 2
max_reconnect_attempts = -1
@@ -257,20 +311,11 @@ 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
# dedupe by payload
# --- dedupe by exact payload bytes ---
digest = hashlib.blake2b(payload, digest_size=16).digest()
nowm = monotonic()
exp = recent_payloads.get(digest)
@@ -283,7 +328,7 @@ async def main():
if recent_payloads.get(k, 0) <= cutoff:
recent_payloads.pop(k, None)
# parse payload
# parse
product = mac = fw = "-"
try:
text = payload.decode("utf-8", errors="replace")
@@ -292,24 +337,20 @@ async def main():
except Exception:
pass
# optional NetBox lookup for fox100
# optional NetBox lookup for fox100 (inline, sequential)
host_suffix = ""
if nb_enabled and product == "fox100":
if 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,
)
host = 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
except Exception as e:
await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
# print one line
# print one compact line
line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}"
if args.include_subject:
line += f" subject={msg.subject}"
@@ -324,7 +365,8 @@ async def main():
await nc.subscribe(args.subject, cb=message_handler)
await log_status(
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'}"
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"
)
# graceful shutdown
@@ -346,8 +388,7 @@ async def main():
await nc.drain()
finally:
await nc.close()
if session:
await session.close()
if __name__ == "__main__":
try: