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

1458

Browse Source
This commit is contained in:
pavel
2026-02-11 14:58:11 +02:00
parent 5d19eb0f26
commit 5c03bce9cc
1 changed files with 38 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
files/nats_registration_listener.py
View File

@@ -8,7 +8,8 @@ NATS Registration Listener (fox100 + NetBox hostname/action_next lookup + timing
* 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)
(and publish to RabbitMQ)
* After successful publish (HTTP 200 + routed true): copy action_next -> action_last in NetBox (PATCH)
* If action_next present -> prepend 3x ASCII BEL to stdout line (kept behavior)
"""
@@ -41,7 +42,6 @@ NB_TIMEOUT = 3.0 # seconds per HTTP GET
# =========================
# 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
@@ -185,6 +185,27 @@ def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] =
return None, 597
# NetBox PATCH helper (JSON in/out)
def http_patch_json(url: str, payload_obj: Dict[str, Any], headers: Optional[Dict[str, str]] = None, timeout: float = NB_TIMEOUT):
body = json.dumps(payload_obj).encode("utf-8")
h = dict(headers or {})
h["Content-Type"] = "application/json"
req = Request(url, data=body, headers=h, method="PATCH")
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
@@ -401,7 +422,7 @@ async def main():
f" publish_body: {json.dumps(rmq_body, ensure_ascii=False)}"
)
# Publish disabled (kept in place, commented out)
# Publish
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}")
@@ -413,7 +434,20 @@ async def main():
routed = False
if not routed:
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
else:
# Copy action_next -> action_last in NetBox (surgical)
try:
base = NB_URL.rstrip("/")
nb_headers = {
"Accept": "application/json",
"Authorization": f"Token {NB_TOKEN}",
}
patch_body = {"custom_fields": {"action_last": action_next_str}}
_, pcode = http_patch_json(f"{base}/api/dcim/devices/{dev_id}/", patch_body, headers=nb_headers, timeout=NB_TIMEOUT)
if pcode != 200:
await log_status(f"[{ts()}] nb: action_last patch http={pcode} host={host} dev_id={dev_id}")
except Exception as e:
await log_status(f"[{ts()}] nb: action_last patch error host={host!r} dev_id={dev_id!r} err={e!r}")
# Bell behavior: ring 3x BEL when action_next present
bell_prefix = "\a" * 3