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

0553

Browse Source
This commit is contained in:
pavel
2026-04-23 05:53:29 +03:00
parent 7834b2acfc
commit 3e6dbdcbd4
1 changed files with 885 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

885
files/netbox_subsystem_ikejanum_recursive.py
View File

@@ -0,0 +1,885 @@
#!/usr/bin/env python3
"""
Recursive single-target sync: Subsystem -> NetBox (Source of Truth), with depth.
Usage:
./netbox_subsystem_ikejanum <device_name> [--deep N]
Behavior:
- Start with <device_name>.
- 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 "<none>"
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())