16:54

files/nats_registration_listener.py

@@ -1,17 +1,10 @@
 #!/usr/bin/env python3
 """
-NATS Registration Listener (compact one-line output + strict read-only dedupe)
+NATS Registration Listener (fox100 + NetBox hostname/tag lookup + timing + problem counter)
-------------------------------------------------------------------------------
+-------------------------------------------------------------------------------------------
-Print exactly one line per message:
+- One device GET (status + tags), no duplicate fetch
-  [TS] product=<productName> mac=<mac> fw=<active>
+- On tag match (and status=active), prepend ASCII BEL to stdout line to alert
-
+- Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting
-- 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.
-
-Install deps:
-  pip install --upgrade nats-py
 """
 
 import argparse
@@ -22,61 +15,83 @@ 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
+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
+import base64  # NEW: for RabbitMQ Basic Auth
 
 import nats
 
 
+# =========================
+# NetBox hardcoded config
+# =========================
+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
+
+# =========================
+# RabbitMQ hardcoded config (immediate publish like rmq-ikeja-pub3.sh without delay)
+# =========================
+RMQ_HOST = "10.210.12.2"
+RMQ_PORT = 15672
+RMQ_USER = "admin"
+RMQ_PASS = "change_me"
+RMQ_VHOST = "app"
+RMQ_EXCHANGE_WORK = "deviceconfig"  # direct exchange (immediate)
+RMQ_ROUTING_KEY = "deviceconfig"
+RMQ_TIMEOUT = 3.0
+
+# Product -> Tag slug mapping (future-proof for fox200 later)
+PRODUCT_TAG_SLUG = {
+    "fox100": "fox100-auto-upgrade-latest",
+    # "fox200": "fox200-auto-upgrade-latest",
+}
+
+# Global counter for any NetBox-related problems
+NB_PROBLEM_COUNTER = 0
+NB_PROBLEM_LOCK = asyncio.Lock()
+
+
+# =========================
+# Arg parsing
+# =========================
 def parse_args():
     p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.")
-    p.add_argument(
+    p.add_argument("--servers", nargs="+", default=["nats://127.0.0.1:4222"])
-        "--servers",
+    p.add_argument("--subject", default="registration")
-        nargs="+",
+    p.add_argument("--queue", default=None)
-        help="One or more NATS server URLs (e.g., nats://127.0.0.1:4222).",
+    p.add_argument("--name", default="registration-listener")
-        default=[os.environ.get("NATS_URL", "nats://127.0.0.1:4222")],
+    p.add_argument("--creds")
-    )
+    p.add_argument("--user")
-    p.add_argument(
+    p.add_argument("--password")
-        "--subject",
+    p.add_argument("--token")
-        help="NATS subject to subscribe to.",
+    p.add_argument("--tls-ca")
-        default=os.environ.get("NATS_SUBJECT", "registration"),
+    p.add_argument("--tls-cert")
-    )
+    p.add_argument("--tls-key")
-    p.add_argument("--queue", help="Optional queue group name.", default=os.environ.get("NATS_QUEUE"))
+    p.add_argument("--insecure", action="store_true")
-    p.add_argument("--name", help="Client connection name.", default="registration-listener")
+    p.add_argument("--include-subject", action="store_true")
-
-    # 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.")
     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
 
 
@@ -85,7 +100,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,61 +109,171 @@ 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"] if isinstance(root.get("data"), dict) else root
-        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", {})
     if isinstance(eths, dict):
         chosen = pick_first_interface(eths)
         if chosen and isinstance(chosen.get("macAddress"), str):
             mac = chosen["macAddress"]
-    except Exception:
-        pass
-
     return product, mac, fw_active
 
 
+# =========================
+# NetBox lookup (urllib)
+# =========================
+def normalize_mac(mac: str) -> Optional[str]:
+    if not mac or not isinstance(mac, str):
+        return None
+    s = mac.strip().lower().replace("-", ":")
+    hex_only = "".join(ch for ch in s if ch in "0123456789abcdef")
+    if len(hex_only) == 12:
+        return ":join".replace(":", "").join([":".join(hex_only[i:i+2] for i in range(0, 12, 2))])  # (keeping original behavior; no change)
+    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
+
+
+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()
+            return json.loads(data.decode("utf-8", errors="replace")), 200
+    except HTTPError as e:
+        return None, getattr(e, "code", 599)
+    except URLError:
+        return None, 598
+    except Exception:
+        return None, 597
+
+
+# NEW: RabbitMQ management API POST helper (basic auth; JSON in/out)
+def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, timeout: float = RMQ_TIMEOUT):
+    body = json.dumps(payload_obj).encode("utf-8")
+    headers = {"Content-Type": "application/json"}
+    if user and password:
+        token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
+        headers["Authorization"] = f"Basic {token}"
+    req = Request(url, data=body, headers=headers, method="POST")
+    try:
+        with urlopen(req, timeout=timeout) as resp:
+            data = resp.read()
+            try:
+                return json.loads(data.decode("utf-8", errors="replace")), resp.status
+            except Exception:
+                return None, resp.status
+    except HTTPError as e:
+        return None, getattr(e, "code", 599)
+    except URLError:
+        return None, 598
+    except Exception:
+        return None, 597
+
+
+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_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, 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, None
+
+    base = NB_URL.rstrip("/")
+    h = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+        "Authorization": f"Token {NB_TOKEN}",
+    }
+
+    # 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:
+        asyncio.create_task(nb_problem(log_status, f"[{ts()}] nb: mac query http={code} mac={mac_norm}"))
+        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, None
+
+    rec = results[0]
+    aot = (rec.get("assigned_object_type") or "").strip()
+    aoid = rec.get("assigned_object_id")
+
+    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, None
+    if aot != "dcim.interface":
+        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, None
+
+    # 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, 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, None
+
+    # Step 3: Device detail (single fetch for 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 info" (no problem increment)
+        return host, aoid, dev_id, None, None
+
+    status_val = ((device.get("status") or {}).get("value")) or None
+    tags = device.get("tags") or []
+    tag_slugs = set()
+    for t in tags:
+        slug = t.get("slug")
+        if isinstance(slug, str):
+            tag_slugs.add(slug)
+
+    return host, aoid, dev_id, (status_val if isinstance(status_val, str) else None), tag_slugs
+
+
+# =========================
+# 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_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
 
     async def disconnected_cb():
         await log_status(f"[{ts()}] Disconnected from NATS.")
@@ -167,8 +291,8 @@ async def main():
         servers=args.servers,
         name=args.name,
         allow_reconnect=True,
-        reconnect_time_wait=reconnect_time_wait,
+        reconnect_time_wait=2,
-        max_reconnect_attempts=max_reconnect_attempts,
+        max_reconnect_attempts=-1,
         disconnected_cb=disconnected_cb,
         reconnected_cb=reconnected_cb,
         error_cb=error_cb,
@@ -181,28 +305,22 @@ async def main():
     )
 
     async def message_handler(msg: nats.aio.msg.Msg):
-        now_s = ts()
+        t_start = monotonic()
         payload = msg.data
 
-        # --- STRICT DEDUPE by exact payload bytes (read-only) ---
+        # strict payload dedupe
-        # Use a stable, compact digest to key the recent map.
         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 list(recent_payloads.keys()):
-            for k in recent_payloads_keys:
+                if recent_payloads[k] <= cutoff:
-                if recent_payloads.get(k, 0) <= cutoff:
                     recent_payloads.pop(k, None)
 
-        # Parse and print one line
         product = mac = fw = "-"
         try:
             text = payload.decode("utf-8", errors="replace")
@@ -211,12 +329,75 @@ async def main():
         except Exception:
             pass
 
-        line = f"[{now_s}] product={product} mac={mac} fw={fw}"
+        host_suffix = ""
+        action_suffix = ""
+        bell_prefix = ""  # ASCII BEL when we have a tag match
+        netbox_time_ms = 0.0
+
+        desired_slug = PRODUCT_TAG_SLUG.get(product)
+        if product == "fox100":
+            nb_start = time.perf_counter()
+            try:
+                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}"
+
+                # Tag-based action (unchanged)
+                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:
+                        bell_prefix = "\a" * 5
+                        action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
+
+                # NEW: Publish immediate RMQ message like rmq-ikeja-pub3.sh (no delay)
+                if host:
+                    try:
+                        rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
+                        rmq_body = {
+                            "properties": {
+                                "content_type": "application/json"
+                            },
+                            "routing_key": RMQ_ROUTING_KEY,
+                            "payload": json.dumps({
+                                "inscope_device": host,
+                                "task_name": "sot-updater-upgradecmd",
+                            }),
+                            "payload_encoding": "string",
+                        }
+                        resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT)
+                        if code != 200:
+                            await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}")
+                        else:
+                            # if response JSON has 'routed' false, note it (rmq-ikeja-pub3.sh warns in that case)
+                            routed = False
+                            try:
+                                routed = bool((resp or {}).get("routed", False))
+                            except Exception:
+                                routed = False
+                            if not routed:
+                                await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
+                    except Exception as e:
+                        await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
+
+            except Exception as e:
+                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
+
+        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}{action_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")
+            # Prepend BEL only when we had a tag match
+            sys.stdout.write(bell_prefix + line + "\n")
             sys.stdout.flush()
 
     if args.queue:
@@ -225,11 +406,9 @@ async def main():
         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"[{ts()}] Listening on subject '{args.subject}' (queue={args.queue or '-'}) via {args.servers} | nb=on"
-        f"| mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled"
     )
 
-    # Graceful shutdown
     stop_event = asyncio.Event()
 
     def handle_signal(*_):
@@ -244,9 +423,7 @@ async def main():
             signal.signal(s, lambda *_: handle_signal())
 
     await stop_event.wait()
-    try:
     await nc.drain()
-    finally:
     await nc.close()