diff --git a/files/netbox_subsystem_ikejanum_recursive.py b/files/netbox_subsystem_ikejanum_recursive.py index e69de29..02d97fa 100644 --- a/files/netbox_subsystem_ikejanum_recursive.py +++ b/files/netbox_subsystem_ikejanum_recursive.py @@ -0,0 +1,885 @@ +#!/usr/bin/env python3 +""" +Recursive single-target sync: Subsystem -> NetBox (Source of Truth), with depth. + +Usage: + ./netbox_subsystem_ikejanum [--deep N] + +Behavior: +- Start with . +- Sync from Subsystem into NetBox (create device if missing). +- If device missing in Subsystem but present in NetBox -> set NetBox status=failed. +- While syncing: + * If an IP is MOVED from another NetBox device -> enqueue that old device for sync (depth+1) + * If a stale IP is being DELETED from a device -> find Subsystem device that owns that IP and enqueue it (depth+1) + +Depth: +- Default depth: 3 +- --deep N allowed, but N is hard-limited to <= 10. + +NEW: +- --info-only: Print the raw Subsystem record (JSON) for the device and exit. + No NetBox changes, no recursion. + +Exit codes: + 0 success + 1 runtime error / not found (info-only) + 2 config/arg error +""" + +# ========================= +# USER CONFIG — EDIT HERE +# ========================= +NB_URL = "http://netbox.gt-tiso.ikeja.co.za" +NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" + +SUBSYSTEM_BASE = "https://subsystem.ikeja.co.za" +SUBSYSTEM_TOKEN = "HJL+&XRCoeHwh5?13@gxvg86qD#kQfgc" + +ROLE_CPE = 1 +TYPE_FOX100_CPE = 1 +TYPE_FOX200 = 2 +SITE_ID = 1 + +DEFAULT_MIN_LAST_DETECTED_MINUTES = 0 # 0 = accept any age +DEFAULT_DEEP = 3 +MAX_DEEP = 10 +# ========================= + +import os +import sys +import json +import time +import logging +import argparse +from typing import Optional, Dict, Any, Tuple, List, Set, Deque +from collections import deque + +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime + +# ------------- Logging ------------- +def setup_logging(troubleshoot: bool) -> None: + level = logging.DEBUG if troubleshoot else logging.INFO + logging.basicConfig( + level=level, + format='%(asctime)s %(levelname)s %(message)s', + datefmt='%H:%M:%S' + ) + +def summarize_body(body: Any, limit: int = 500) -> str: + if body is None: + return "" + if isinstance(body, (dict, list)): + s = json.dumps(body) + else: + s = str(body) + if len(s) > limit: + return s[:limit] + f"... (+{len(s)-limit}B)" + return s + +def is_ascii_or_die(label: str, value: str) -> None: + try: + value.encode('latin-1') + except UnicodeEncodeError: + bad = ''.join(ch for ch in value if ord(ch) > 127) + logging.error( + "%s contains non-ASCII characters (e.g. %r). Please paste the exact token without smart punctuation.", + label, bad, + ) + raise SystemExit(2) + +# ------------- HTTP Session helpers ------------- +class Http: + def __init__(self, troubleshoot: bool = False): + self.s_nb = self._new_session() + self.s_subsystem = self._new_session() + self.troubleshoot = troubleshoot + + def _new_session(self) -> requests.Session: + s = requests.Session() + retries = Retry( + total=3, + connect=3, + read=3, + status=3, + backoff_factor=0.75, + status_forcelist=(429, 500, 502, 503, 504), + allowed_methods=("GET", "POST", "PATCH", "PUT", "DELETE"), + respect_retry_after_header=True, + ) + adapter = HTTPAdapter(max_retries=retries, pool_connections=16, pool_maxsize=32) + s.mount('http://', adapter) + s.mount('https://', adapter) + s.headers['Accept'] = 'application/json' + s.headers['Content-Type'] = 'application/json' + return s + + def nb(self, method: str, url: str, token: str, **kw) -> requests.Response: + headers = kw.pop('headers', {}) + headers['Authorization'] = f'Token {token}' + t0 = time.time() + resp = self.s_nb.request(method, url, headers=headers, timeout=30, **kw) + dt = (time.time() - t0) * 1000 + if self.troubleshoot or resp.status_code >= 400: + logging.debug( + "NB %s %s [%s] %.1fms\n req:%s\n resp:%s", + method, url, resp.status_code, dt, + summarize_body(kw.get('data') or kw.get('json')), + summarize_body(resp.text), + ) + return resp + + def subsystem(self, method: str, url: str, token: str, **kw) -> requests.Response: + headers = kw.pop('headers', {}) + headers['token'] = token + t0 = time.time() + resp = self.s_subsystem.request(method, url, headers=headers, timeout=30, **kw) + dt = (time.time() - t0) * 1000 + if self.troubleshoot or resp.status_code >= 400: + logging.debug( + "SUBSYS %s %s [%s] %.1fms\n resp:%s", + method, url, resp.status_code, dt, + summarize_body(resp.text), + ) + return resp + +# ------------- Utility ------------- +def normalize_mac_from_subsystem(mac_raw: str) -> Optional[str]: + if not mac_raw: + return None + s = mac_raw.strip().replace(":", "").replace("-", "") + s = s.upper() + try: + chunks = [s[i:i+2] for i in range(0, len(s), 2)] + return ":".join(chunks) + except Exception: + logging.warning(" !! failed to normalize MAC from subsystem: %r", mac_raw) + return None + +def parse_last_detected(ts: Optional[str]) -> Optional[datetime]: + if not ts: + return None + try: + dt = parsedate_to_datetime(ts) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + else: + dt = dt.astimezone(timezone.utc) + return dt + except Exception: + logging.warning(" !! cannot parse last_detected: %r", ts) + return None + +def minutes_age(now_utc: datetime, past_utc: datetime) -> float: + return (now_utc - past_utc).total_seconds() / 60.0 + +def ip_strip_prefix(ip_with_prefix: str) -> str: + # "10.0.0.1/32" -> "10.0.0.1" + return (ip_with_prefix or "").split("/")[0].strip() + +# ------------- Subsystem client (with caching) ------------- +class Subsystem: + def __init__(self, base: str, token: str, http: Http): + self.base = base.rstrip('/') + self.token = token + self.http = http + self._cache_devices: Optional[list[Dict[str, Any]]] = None + + def fetch_outdoor_devices(self) -> list[Dict[str, Any]]: + url = f"{self.base}/customers/wave-devices/get-outdoor-devices" + r = self.http.subsystem("POST", url, self.token) + r.raise_for_status() + data = r.json() + return data.get("device_list") or [] + + def _ensure_cache(self) -> None: + if self._cache_devices is None: + self._cache_devices = self.fetch_outdoor_devices() + + def find_device_by_name(self, target_name: str) -> Optional[Dict[str, Any]]: + self._ensure_cache() + assert self._cache_devices is not None + for rec in self._cache_devices: + name = (rec.get("name") or "").strip() + if name.lower() == target_name.lower(): + return rec + return None + + def find_device_by_ip(self, ip: str) -> Optional[Dict[str, Any]]: + if not ip: + return None + self._ensure_cache() + assert self._cache_devices is not None + ip = ip.strip() + for rec in self._cache_devices: + rip = (rec.get("ip") or "").strip() + if rip == ip: + return rec + return None + +# ------------- NetBox API wrappers ------------- +class NetBox: + def __init__( + self, + base: str, + token: str, + role_cpe: int, + type_fox100: int, + type_fox200: int, + site_id: int, + http: Http, + dry_run: bool = False, + ): + self.base = base.rstrip('/') + self.token = token + self.role_cpe = role_cpe + self.type_fox100 = type_fox100 + self.type_fox200 = type_fox200 + self.site_id = site_id + self.http = http + self.dry = dry_run + self.allow_moves = True + + def _url(self, path: str) -> str: + return f"{self.base}{path}" + + def _req(self, method: str, path: str, **kw) -> requests.Response: + if self.dry and method in ("POST", "PATCH", "PUT", "DELETE"): + logging.info("DRY %s %s", method, path) + r = requests.Response() + if method == "POST": + r.status_code = 201 + r._content = b'{"id": 0}' + else: + r.status_code = 200 + r._content = b'{}' + r.headers['Content-Type'] = 'application/json' + return r + return self.http.nb(method, self._url(path), self.token, **kw) + + # --- devices --- + def get_device_id_by_name(self, name: str) -> Optional[int]: + r = self._req("GET", "/api/dcim/devices/", params={"name": name}) + r.raise_for_status() + data = r.json() + if data.get('results'): + return data['results'][0]['id'] + return None + + def get_device(self, dev_id: int) -> Dict[str, Any]: + r = self._req("GET", f"/api/dcim/devices/{dev_id}/") + r.raise_for_status() + return r.json() + + def get_device_name(self, dev_id: int) -> Optional[str]: + r = self._req("GET", f"/api/dcim/devices/{dev_id}/") + if r.status_code == 200: + return r.json().get("name") + return None + + def create_device(self, name: str, dtype_id: int) -> Optional[int]: + payload = { + "name": name, + "role": self.role_cpe, + "device_type": dtype_id, + "site": self.site_id, + "status": "active", + } + r = self._req("POST", "/api/dcim/devices/", json=payload) + if r.status_code == 201: + dev_id = r.json().get('id') + logging.info(" -> created NetBox device name=%s id=%s", name, dev_id) + return dev_id + logging.error("device create failed (%s) HTTP=%s body=%s", name, r.status_code, summarize_body(r.text)) + return None + + def patch_device(self, dev_id: int, patch: Dict[str, Any]) -> bool: + if not patch: + return True + r = self._req("PATCH", f"/api/dcim/devices/{dev_id}/", json=patch) + if 200 <= r.status_code < 300: + return True + logging.error("device patch failed dev=%s HTTP=%s body=%s", dev_id, r.status_code, summarize_body(r.text)) + return False + + def set_primary_ip4_if_changed(self, dev: Dict[str, Any], ip_id: int) -> bool: + curr = (dev.get('primary_ip4') or {}).get('id') + if curr == ip_id: + return True + r = self._req("PATCH", f"/api/dcim/devices/{dev['id']}/", json={"primary_ip4": ip_id}) + ok = 200 <= r.status_code < 300 + if not ok: + logging.error("set primary_ip4 failed dev=%s HTTP=%s body=%s", dev['id'], r.status_code, summarize_body(r.text)) + return ok + + def set_status(self, dev_id: int, status: str) -> bool: + r = self._req("PATCH", f"/api/dcim/devices/{dev_id}/", json={"status": status}) + if 200 <= r.status_code < 300: + return True + logging.error("status patch failed dev=%s status=%s HTTP=%s body=%s", dev_id, status, r.status_code, summarize_body(r.text)) + return False + + def add_journal_entry(self, dev_id: int, comments: str, kind: str = "info") -> bool: + payload = { + "assigned_object_type": "dcim.device", + "assigned_object_id": dev_id, + "kind": kind, + "comments": comments, + } + r = self._req("POST", "/api/extras/journal-entries/", json=payload) + if 200 <= r.status_code < 300: + return True + logging.error( + "journal entry create failed dev=%s kind=%s HTTP=%s body=%s", + dev_id, kind, r.status_code, summarize_body(r.text) + ) + return False + + # --- interfaces --- + def get_iface_id(self, dev_id: int, name: str) -> Optional[int]: + r = self._req("GET", "/api/dcim/interfaces/", params={"device_id": dev_id, "name": name}) + r.raise_for_status() + data = r.json() + if data.get('results'): + return data['results'][0]['id'] + return None + + def ensure_eth0(self, dev_id: int) -> Optional[int]: + ifid = self.get_iface_id(dev_id, 'eth0') + if ifid: + return ifid + payload = {"device": dev_id, "name": "eth0", "type": "1000base-t"} + r = self._req("POST", "/api/dcim/interfaces/", json=payload) + if r.status_code == 201: + return r.json()['id'] + if r.status_code == 400: + return self.get_iface_id(dev_id, 'eth0') + logging.error("interface create failed device=%s HTTP=%s body=%s", dev_id, r.status_code, summarize_body(r.text)) + return None + + def get_device_id_of_interface(self, iface_id: int) -> Optional[int]: + r = self._req("GET", f"/api/dcim/interfaces/{iface_id}/") + if r.status_code == 200: + return (r.json().get('device') or {}).get('id') + return None + + def get_interface_detail(self, iface_id: int) -> Optional[Dict[str, Any]]: + r = self._req("GET", f"/api/dcim/interfaces/{iface_id}/") + if 200 <= r.status_code < 300: + return r.json() + logging.error("interface fetch failed iface=%s HTTP=%s body=%s", iface_id, r.status_code, summarize_body(r.text)) + return None + + # --- IP inventory helpers --- + def list_device_ips(self, dev_id: int) -> list[Dict[str, Any]]: + r = self._req("GET", "/api/ipam/ip-addresses/", params={"device_id": dev_id, "limit": 1000}) + if r.status_code == 200: + return r.json().get('results') or [] + # fallback by interface + ips: list[Dict[str, Any]] = [] + r2 = self._req("GET", "/api/dcim/interfaces/", params={"device_id": dev_id, "limit": 1000}) + if r2.status_code == 200: + for iface in (r2.json().get('results') or []): + ifid = iface.get('id') + r3 = self._req("GET", "/api/ipam/ip-addresses/", params={ + "assigned_object_type": "dcim.interface", + "assigned_object_id": ifid, + "limit": 1000, + }) + if r3.status_code == 200: + ips.extend(r3.json().get('results') or []) + return ips + + def prune_other_ips_with_details(self, dev_id: int, keep_ip_id: int) -> List[Tuple[int, str]]: + """ + Delete all other IPs assigned to this device, keeping only keep_ip_id. + Returns list of (ip_id, address) that were removed. + """ + removed: List[Tuple[int, str]] = [] + all_ips = self.list_device_ips(dev_id) + for rec in all_ips: + rid = rec.get('id') + if rid == keep_ip_id: + continue + addr = rec.get('address') + logging.warning(" -> removing stale IP %s (id=%s) from device %s", addr, rid, dev_id) + self._req("DELETE", f"/api/ipam/ip-addresses/{rid}/") + if rid is not None and addr: + removed.append((int(rid), str(addr))) + return removed + + # --- MAC helpers --- + def ensure_single_mac_on_iface(self, iface_id: int, mac_norm: str) -> Tuple[int, int]: + created = 0 + deleted = 0 + iface = self.get_interface_detail(iface_id) + if not iface: + logging.error(" !! cannot fetch interface detail for MAC sync (iface=%s)", iface_id) + return (0, 0) + + macs = iface.get("mac_addresses") or [] + matching_ids: list[int] = [] + bad_ids: list[int] = [] + + for m in macs: + mid = m.get("id") + mval = (m.get("mac_address") or "").upper() + if mval == mac_norm: + if mid is not None: + matching_ids.append(mid) + else: + if mid is not None: + bad_ids.append(mid) + + if not matching_ids: + payload = { + "mac_address": mac_norm, + "assigned_object_type": "dcim.interface", + "assigned_object_id": iface_id, + } + logging.info(" -> create MAC %s on iface %s", mac_norm, iface_id) + r = self._req("POST", "/api/dcim/mac-addresses/", json=payload) + if r.status_code == 201: + created += 1 + else: + logging.error(" !! MAC create failed iface=%s mac=%s HTTP=%s body=%s", iface_id, mac_norm, r.status_code, summarize_body(r.text)) + else: + extra_ids = matching_ids[1:] + bad_ids.extend([i for i in extra_ids if i is not None]) + + for mid in bad_ids: + logging.warning(" -> removing stale/duplicate MAC entry id=%s from iface=%s", mid, iface_id) + self._req("DELETE", f"/api/dcim/mac-addresses/{mid}/") + deleted += 1 + + return (created, deleted) + + # --- IP addresses --- + def get_ip_by_address(self, addr: str) -> Dict[str, Any] | None: + r = self._req("GET", "/api/ipam/ip-addresses/", params={"address": f"{addr}/32"}) + r.raise_for_status() + res = r.json().get('results') or [] + return res[0] if res else None + + def assign_ip_to_iface(self, ip_id: int, iface_id: int) -> bool: + r = self._req("PATCH", f"/api/ipam/ip-addresses/{ip_id}/", json={ + "assigned_object_type": "dcim.interface", + "assigned_object_id": iface_id, + }) + return r.status_code == 200 + + def ensure_ip_for_device(self, dev_id: int, iface_id: int, addr: str) -> Tuple[Optional[int], str, Optional[int]]: + """ + Ensure addr/32 exists and is assigned to iface_id on dev_id. + Returns (ip_id, action, old_device_id_if_moved) + action: created, reused, assigned, moved, skipped, error + """ + ip_rec = self.get_ip_by_address(addr) + if not ip_rec: + payload = { + "address": f"{addr}/32", + "status": "active", + "assigned_object_type": "dcim.interface", + "assigned_object_id": iface_id, + } + r = self._req("POST", "/api/ipam/ip-addresses/", json=payload) + if r.status_code == 201: + return (r.json()['id'], "created", None) + logging.error("ip create failed iface=%s addr=%s HTTP=%s body=%s", iface_id, addr, r.status_code, summarize_body(r.text)) + return (None, "error", None) + + ip_id = ip_rec['id'] + aot = ip_rec.get('assigned_object_type') or '' + if not aot: + ok = self.assign_ip_to_iface(ip_id, iface_id) + return (ip_id if ok else None, "assigned" if ok else "error", None) + + if aot == 'dcim.interface': + assigned_ifid = ip_rec.get('assigned_object_id') + if str(assigned_ifid) == str(iface_id): + return (ip_id, "reused", None) + + assigned_dev = self.get_device_id_of_interface(assigned_ifid) if assigned_ifid else None + if assigned_dev and str(assigned_dev) == str(dev_id): + logging.info(" -> IP %s already on same device (iface=%s); reusing.", addr, assigned_ifid) + return (ip_id, "reused", None) + + if self.allow_moves: + old_dev_id = assigned_dev + + # clear primary_ip4 on old device if needed + if old_dev_id: + r_old = self._req("GET", f"/api/dcim/devices/{old_dev_id}/") + if r_old.status_code == 200: + old_primary_id = (r_old.json().get("primary_ip4") or {}).get("id") + if str(old_primary_id) == str(ip_id): + r_clr = self._req("PATCH", f"/api/dcim/devices/{old_dev_id}/", json={"primary_ip4": None}) + if not (200 <= r_clr.status_code < 300): + logging.error(" -> cannot clear old device %s primary_ip4 for IP %s; aborting move", old_dev_id, addr) + return (None, "error", None) + + logging.warning( + " -> IP %s currently belongs to device %s (iface %s); moving to this device (iface %s)", + addr, old_dev_id, assigned_ifid, iface_id, + ) + r2 = self._req("PATCH", f"/api/ipam/ip-addresses/{ip_id}/", json={ + "assigned_object_type": "dcim.interface", + "assigned_object_id": iface_id, + }) + if r2.status_code == 200: + return (ip_id, "moved", old_dev_id) + logging.error(" -> move failed for IP %s HTTP=%s body=%s", addr, r2.status_code, summarize_body(r2.text)) + return (None, "error", None) + + return (None, "skipped", None) + + logging.warning(" -> IP %s assigned to %s; skipping.", addr, aot) + return (None, "skipped", None) + +# ------------- Core sync functions ------------- +def sync_from_subsystem_record( + target: str, + rec: Dict[str, Any], + nb: NetBox, + subsystem: Subsystem, + now_utc: datetime, + min_last_detected_minutes: int, + enqueue_fn, +) -> Dict[str, Any]: + """ + enqueue_fn(name: str, reason: str) -> None + """ + summary = { + "target": target, + "subsystem_found": True, + "netbox_created": False, + "status_set_active": False, + "node_updated": False, + "mac_created": 0, + "mac_deleted": 0, + "ip_action": "none", + "ips_pruned": 0, + "primary_set": False, + "skipped_stale": False, + "errors": 0, + } + + name = rec.get("name") or target + ip = rec.get("ip") + mac_raw = rec.get("mac") + connected_node = rec.get("connected_node") + last_detected = rec.get("last_detected") + + dt_last = parse_last_detected(last_detected) + if min_last_detected_minutes > 0: + if dt_last is None: + logging.info(" -> skip: last_detected unavailable; requires <= %s minutes", min_last_detected_minutes) + summary["skipped_stale"] = True + return summary + age_min = minutes_age(now_utc, dt_last) + if age_min > min_last_detected_minutes: + logging.info(" -> skip: last_detected age %.1f min > allowed %s min", age_min, min_last_detected_minutes) + summary["skipped_stale"] = True + return summary + + dev_id = nb.get_device_id_by_name(name) + if not dev_id: + logging.info(" -> device not found in NetBox; creating") + dev_id = nb.create_device(name, nb.type_fox100) + if not dev_id: + summary["errors"] += 1 + return summary + summary["netbox_created"] = True + + dev = nb.get_device(dev_id) + curr_status = dev.get("status", {}).get("value") if isinstance(dev.get("status"), dict) else dev.get("status") + if curr_status != "active": + active_reason = "device present in subsystem" + if min_last_detected_minutes > 0: + active_reason += f" and last_detected within {min_last_detected_minutes} minutes" + else: + active_reason += " and freshness filter accepts it" + if nb.set_status(dev_id, "active"): + nb.add_journal_entry(dev_id, f"setting to active because {active_reason}", kind="info") + summary["status_set_active"] = True + dev = nb.get_device(dev_id) + else: + summary["errors"] += 1 + + cf = dev.get("custom_fields") or {} + current_node = (cf.get("nodeName") if isinstance(cf, dict) else None) + + patch: Dict[str, Any] = {} + if connected_node and connected_node != current_node: + patch.setdefault("custom_fields", {})["nodeName"] = connected_node + logging.info(" -> patch custom_fields.nodeName: %r -> %r", current_node, connected_node) + if nb.patch_device(dev_id, patch): + summary["node_updated"] = True + dev = nb.get_device(dev_id) + + iface_id = nb.ensure_eth0(dev_id) + if not iface_id: + logging.error(" !! interface ensure/create failed for device=%s", dev_id) + summary["errors"] += 1 + return summary + + mac_norm = normalize_mac_from_subsystem(mac_raw) if mac_raw else None + if mac_norm: + logging.info(" -> ensure single MAC %s on iface %s", mac_norm, iface_id) + c, d = nb.ensure_single_mac_on_iface(iface_id, mac_norm) + summary["mac_created"] += c + summary["mac_deleted"] += d + else: + logging.info(" -> MAC missing/unusable from subsystem; skipping MAC sync") + + if ip and str(ip).lower() != "null" and ip != "0.0.0.0": + logging.info(" -> ensure IP %s for device %s (iface %s)", ip, dev_id, iface_id) + ip_id, ip_action, old_dev_id = nb.ensure_ip_for_device(dev_id, iface_id, ip) + summary["ip_action"] = ip_action + + # If we moved from some other NetBox device, enqueue it for repair + if ip_action == "moved" and old_dev_id: + old_name = nb.get_device_name(old_dev_id) + if old_name and old_name.lower() != name.lower(): + enqueue_fn(old_name, f"ip_moved_away:{ip}") + + if ip_id: + removed = nb.prune_other_ips_with_details(dev_id, ip_id) + summary["ips_pruned"] = len(removed) + + # For each deleted stale IP, see who owns that IP in Subsystem and enqueue them. + for _rid, addr_pref in removed: + stale_ip = ip_strip_prefix(addr_pref) + rec2 = subsystem.find_device_by_ip(stale_ip) + if rec2: + n2 = rec2.get("name") + if n2 and n2.lower() != name.lower(): + enqueue_fn(n2, f"stale_ip_deleted:{stale_ip}") + + dev = nb.get_device(dev_id) + before = (dev.get("primary_ip4") or {}).get("id") + if nb.set_primary_ip4_if_changed(dev, ip_id): + if str(before) != str(ip_id): + summary["primary_set"] = True + else: + summary["errors"] += 1 + else: + logging.info(" -> IP missing or 0.0.0.0; skipping IP sync") + + return summary + +def mark_failed_if_in_netbox_only(target: str, nb: NetBox) -> Dict[str, Any]: + summary = { + "target": target, + "subsystem_found": False, + "netbox_exists": False, + "status_set_failed": False, + "errors": 0, + } + dev_id = nb.get_device_id_by_name(target) + if not dev_id: + return summary + summary["netbox_exists"] = True + + dev = nb.get_device(dev_id) + curr_status = dev.get("status", {}).get("value") if isinstance(dev.get("status"), dict) else dev.get("status") + if curr_status == "failed": + return summary + + failed_reason = "device missing in subsystem" + if nb.set_status(dev_id, "failed"): + nb.add_journal_entry(dev_id, f"setting to failed, because {failed_reason}", kind="warning") + summary["status_set_failed"] = True + else: + summary["errors"] += 1 + return summary + +def one_liner_single(result: Dict[str, Any]) -> str: + t = result.get("target", "?") + if result.get("subsystem_found") is False: + if result.get("netbox_exists"): + if result.get("status_set_failed"): + return f"{t}: not in subsystem -> NetBox status set to failed" + return f"{t}: not in subsystem -> NetBox status NOT changed (error)" + return f"{t}: not in subsystem and not in NetBox -> nothing to do" + + if result.get("skipped_stale"): + return f"{t}: subsystem record skipped due to last_detected freshness filter" + + parts = [] + if result.get("netbox_created"): + parts.append("created in NetBox") + if result.get("status_set_active"): + parts.append("status set to active") + if result.get("node_updated"): + parts.append("nodeName updated") + if (result.get("mac_created", 0) or 0) > 0 or (result.get("mac_deleted", 0) or 0) > 0: + parts.append(f"mac c{result.get('mac_created',0)}/d{result.get('mac_deleted',0)}") + ia = result.get("ip_action") + if ia and ia != "none": + parts.append(f"ip {ia}") + if (result.get("ips_pruned", 0) or 0) > 0: + parts.append(f"ips pruned={result.get('ips_pruned')}") + if result.get("primary_set"): + parts.append("primary_ip4 set") + if (result.get("errors") or 0) > 0: + parts.append(f"errors={result.get('errors')}") + + if not parts: + return f"{t}: already in sync (no changes)" + return f"{t}: " + ", ".join(parts) + +# ------------- Depth driver ------------- +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("device_name", help="Exact device name to sync (e.g. ikeja12345)") + ap.add_argument( + "--info-only", + action="store_true", + default=False, + help="Print Subsystem record for the device and exit (no NetBox changes)", + ) + ap.add_argument( + "--min-last-detected-minutes", + type=int, + default=DEFAULT_MIN_LAST_DETECTED_MINUTES, + help="Require subsystem last_detected within N minutes (0 = accept any age)", + ) + ap.add_argument( + "--deep", + type=int, + default=DEFAULT_DEEP, + help=f"Recursive depth (default {DEFAULT_DEEP}, max {MAX_DEEP})", + ) + ap.add_argument("--troubleshoot", action='store_true', default=bool(os.getenv('TROUBLESHOOT'))) + ap.add_argument("--dry-run", action='store_true', default=False) + args = ap.parse_args() + + setup_logging(args.troubleshoot) + + if not NB_TOKEN: + logging.error("Set NB_TOKEN in script") + return 2 + if not SUBSYSTEM_TOKEN: + logging.error("Set SUBSYSTEM_TOKEN in script") + return 2 + + is_ascii_or_die("NB_TOKEN", NB_TOKEN) + is_ascii_or_die("SUBSYSTEM_TOKEN", SUBSYSTEM_TOKEN) + + min_min = max(0, int(args.min_last_detected_minutes)) + deep = int(args.deep) + if deep < 1: + deep = 1 + if deep > MAX_DEEP: + logging.warning("[WARN] --deep %s requested, capping to %s", deep, MAX_DEEP) + deep = MAX_DEEP + + http = Http(troubleshoot=args.troubleshoot) + nb = NetBox( + NB_URL, + NB_TOKEN, + ROLE_CPE, + TYPE_FOX100_CPE, + TYPE_FOX200, + SITE_ID, + http, + dry_run=args.dry_run, + ) + subsystem = Subsystem(SUBSYSTEM_BASE, SUBSYSTEM_TOKEN, http) + + # --info-only mode (no NetBox changes, no recursion) + if args.info_only: + rec = subsystem.find_device_by_name(args.device_name.strip()) + if not rec: + print(f"NOT_FOUND {args.device_name.strip()} in Subsystem") + return 1 + print(json.dumps(rec, indent=2, sort_keys=True)) + return 0 + + now_utc = datetime.now(timezone.utc) + + # BFS queue: (name, depth, reason) + q: Deque[Tuple[str, int, str]] = deque() + seen: Set[str] = set() + + root = args.device_name.strip() + q.append((root, 0, "root")) + seen.add(root.lower()) + + totals = { + "processed": 0, + "subsystem_found": 0, + "netbox_failed_set": 0, + "created": 0, + "errors": 0, + "enqueued": 0, + } + + def enqueue(name: str, reason: str, parent_depth: int = 0) -> None: + nonlocal q, seen, totals, deep + if not name: + return + k = name.lower() + if k in seen: + return + next_depth = parent_depth + 1 + if next_depth > deep: + return + seen.add(k) + q.append((name, next_depth, reason)) + totals["enqueued"] += 1 + logging.info(" -> enqueue depth=%s name=%s reason=%s", next_depth, name, reason) + + per_device_one_liners: List[str] = [] + + try: + while q: + name, d, reason = q.popleft() + totals["processed"] += 1 + logging.info("[INFO] depth=%s name=%s reason=%s", d, name, reason) + + rec = subsystem.find_device_by_name(name) + if rec: + totals["subsystem_found"] += 1 + + def enq_child(child_name: str, child_reason: str) -> None: + enqueue(child_name, child_reason, parent_depth=d) + + res = sync_from_subsystem_record( + name, rec, nb, subsystem, now_utc, min_min, enq_child + ) + if res.get("netbox_created"): + totals["created"] += 1 + if (res.get("errors") or 0) > 0: + totals["errors"] += 1 + per_device_one_liners.append(one_liner_single(res)) + else: + res2 = mark_failed_if_in_netbox_only(name, nb) + if res2.get("status_set_failed"): + totals["netbox_failed_set"] += 1 + if (res2.get("errors") or 0) > 0: + totals["errors"] += 1 + per_device_one_liners.append(one_liner_single(res2)) + + except Exception as e: + logging.exception("Unhandled exception in deep sync: %s", e) + print(f"{root}: error (exception during deep sync)") + return 1 + + for line in per_device_one_liners: + print(line) + + print( + f"SUMMARY: root={root} processed={totals['processed']} " + f"subsys_found={totals['subsystem_found']} created={totals['created']} " + f"failed_set={totals['netbox_failed_set']} enqueued={totals['enqueued']} errors={totals['errors']}" + ) + return 0 + +if __name__ == "__main__": + sys.exit(main())