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

1435

Browse Source
This commit is contained in:
pavel
2026-02-11 14:35:50 +02:00
parent c2a8d3fc44
commit 46ea4359ed
1 changed files with 55 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
files/nats_registration_listener.py
View File

@@ -1,14 +1,15 @@
#!/usr/bin/env python3
"""
NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing + problem counter)
-----------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------
- One device GET (status + tags + custom_fields.action_next), no duplicate fetch
- If fox100 and action_next present -> prepend 3x ASCII BEL to stdout line
- RabbitMQ publish logic is DISABLED for now:
* We still PREPARE the RabbitMQ message body
* We clearly log: "ok, here i will execute <content of rabbitmq message>"
* But we DO NOT actually POST it to RabbitMQ (publish call commented out)
- Keeps: nb_problems counter, timings, iface_id diagnostics, same formatting
- For fox100:
* If action_next is empty/absent -> print a simple stdout note and do nothing else
* If action_next is present -> PREPARE RabbitMQ /publish body and log:
"ok, here i will execute <body + decoded payload>"
(but DO NOT actually publish; publish call remains commented out)
* If action_next present -> prepend 3x ASCII BEL to stdout line (kept behavior)
"""
import argparse
@@ -39,7 +40,8 @@ NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" # keep as provided
NB_TIMEOUT = 3.0 # seconds per HTTP GET
# =========================
# RabbitMQ hardcoded config (publish currently disabled)
# RabbitMQ hardcoded config (immediate publish like rmq-ikeja-pub3.sh without delay)
# NOTE: publish is currently disabled in message_handler (kept config unchanged).
# =========================
RMQ_HOST = "10.210.12.2"
RMQ_PORT = 15672
@@ -130,29 +132,19 @@ def extract_fields(obj: Dict[str, Any]):
# NetBox lookup (urllib)
# =========================
def normalize_mac(mac: str) -> Optional[str]:
"""
Normalize MAC to lowercase colon-separated hex pairs: aa:bb:cc:dd:ee:ff
Accepts common formats: aa-bb-cc-dd-ee-ff, aabbccddeeff, aa:bb:...
"""
if not mac or not isinstance(mac, str):
return None
s = mac.strip().lower()
# Strip all non-hex chars to handle aabb.ccdd.eeff, aa-bb-..., etc.
s = mac.strip().lower().replace("-", ":")
hex_only = "".join(ch for ch in s if ch in "0123456789abcdef")
if len(hex_only) == 12:
return ":".join(hex_only[i:i + 2] for i in range(0, 12, 2))
# If it's already colon-separated, validate it strictly.
parts = s.replace("-", ":").split(":")
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 ":".join(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):
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")
@@ -171,9 +163,7 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
# RabbitMQ management API POST helper (basic auth; JSON in/out)
# NOTE: currently not used because publishing is disabled, but left in place for later.
def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None,
timeout: float = RMQ_TIMEOUT):
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:
@@ -203,9 +193,7 @@ async def nb_problem(log_status, msg: str):
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]], Optional[Any]
]:
def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str], Optional[Set[str]], Optional[Any]]:
"""
Resolve MAC -> (hostname, iface_id, device_id, device_status_value, tag_slugs_set, action_next)
- Logs problems for anomalies (mac not found, unassigned, wrong type, iface fetch fail).
@@ -223,8 +211,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
}
# Step 1: MAC lookup
data, code = http_get_json(f"{base}/api/dcim/mac-addresses/", params={"mac_address": mac_norm, "limit": "2"},
headers=h)
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, None
@@ -239,9 +226,7 @@ def nb_lookup_device_by_mac(mac: str, log_status) -> Tuple[
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 '-'}")
)
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}"))
@@ -354,11 +339,10 @@ async def main():
pass
host_suffix = ""
action_suffix = "" # kept; no longer used for tag action
action_suffix = "" # kept; not used
bell_prefix = "" # ASCII BEL when action_next present (3x)
netbox_time_ms = 0.0
# Behavior depends on action_next presence only (for fox100)
if product == "fox100":
nb_start = time.perf_counter()
try:
@@ -368,43 +352,56 @@ async def main():
if iface_id and not host:
host_suffix += f" iface_id={iface_id}"
# If host known, prepare RabbitMQ message and log it, but DO NOT publish.
if host:
try:
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
# Determine if action_next is present (non-empty string, or any truthy value)
has_action_next = False
action_next_str = None
try:
if isinstance(action_next, str):
has_action_next = len(action_next.strip()) > 0
action_next_str = action_next.strip()
has_action_next = len(action_next_str) > 0
else:
has_action_next = bool(action_next)
if has_action_next:
action_next_str = str(action_next)
except Exception:
has_action_next = False
action_next_str = None
# Task naming: keep it simple and explicit.
task_name = "sot-updater-actionnext" if has_action_next else "sot-updater"
if not has_action_next:
# User-requested behavior: if no action -> just shoot a message to stdout and we're ok
async with print_lock:
print(f"[{ts()}] no action_next for host={host}", file=sys.stdout, flush=True)
else:
# Prepare RabbitMQ message with task_name taken from action_next (do not keep old task names)
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
rmq_payload_obj = {
payload_obj = {
"inscope_device": host,
"task_name": task_name,
# IMPORTANT: we are not "executing" it here; we only carry it for later scripts.
"action_next": action_next,
"task_name": action_next_str,
}
payload_raw = json.dumps(payload_obj, separators=(",", ":"), ensure_ascii=False)
rmq_body = {
"properties": {
"content_type": "application/json"
},
"routing_key": RMQ_ROUTING_KEY,
"payload": json.dumps(rmq_payload_obj),
"payload": payload_raw,
"payload_encoding": "string",
}
# (b) Explicit log of what would be executed
await log_status(f"[{ts()}] ok, here i will execute {json.dumps(rmq_body, ensure_ascii=False)} url={rmq_url}")
# Visual, easy-to-read log of what would be executed
await log_status(
f"[{ts()}] ok, here i will execute\n"
f" url: {rmq_url}\n"
f" routing_key: {RMQ_ROUTING_KEY}\n"
f" payload_raw: {payload_raw}\n"
f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}"
)
# (c) Publishing is disabled for now (leave code in place, commented out)
# Publish disabled (kept in place, commented out)
# 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}")
@@ -417,13 +414,9 @@ async def main():
# if not routed:
# await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
# Bell behavior: ring 3x BEL only if action_next present
if has_action_next:
# Bell behavior: ring 3x BEL when action_next present
bell_prefix = "\a" * 3
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
@@ -440,7 +433,6 @@ async def main():
line += f" subject={msg.subject}"
async with print_lock:
# Prepend BEL only when we had action_next present (3x)
sys.stdout.write(bell_prefix + line + "\n")
sys.stdout.flush()
@@ -476,4 +468,3 @@ if __name__ == "__main__":
asyncio.run(main())
except KeyboardInterrupt:
pass