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

0823

Browse Source
This commit is contained in:
pavel
2026-03-09 08:23:49 +02:00
parent 2fd1f920c0
commit 7361a85f02
1 changed files with 138 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
files/nats_registration_listener.py
View File

@@ -30,6 +30,7 @@ from urllib.parse import urlencode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
import base64 # for RabbitMQ Basic Auth
from pathlib import Path
import nats
@@ -72,19 +73,13 @@ PRODUCT_TAG_SLUG = {
# "fox200": "fox200-auto-upgrade-latest",
}
# MACs to ignore completely when seen in registrations
IGNORE_MACS_RAW = [
"AA:BB:CC:DD:EE:FF",
"D0:6C:37:00:D5:84",
"44:D1:FA:7E:3F:65",
"D0:6C:37:00:E6:D8",
"C4:93:00:4E:92:7C",
]
# Global counter for any NetBox-related problems
NB_PROBLEM_COUNTER = 0
NB_PROBLEM_LOCK = asyncio.Lock()
# Structured event log output (JSON Lines)
EVENT_LOG_PATH = os.environ.get("EVENT_LOG_PATH", "/opt/containers/nats-registration-listener/logs/registration_events.jsonl")
# =========================
# Arg parsing
@@ -128,6 +123,17 @@ def ts() -> str:
return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z")
def ts_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def append_jsonl(path: str, obj: Dict[str, Any]):
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + "\n")
def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]:
if "eth0" in eths and isinstance(eths["eth0"], dict):
return eths["eth0"]
@@ -167,9 +173,6 @@ def normalize_mac(mac: str) -> Optional[str]:
return None
IGNORE_MACS = {m for m in (normalize_mac(x) for x in IGNORE_MACS_RAW) if m}
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)}"
@@ -376,11 +379,19 @@ async def main():
ssl_ctx = make_ssl_context(args)
print_lock = asyncio.Lock()
event_log_lock = asyncio.Lock()
async def log_status(s: str):
async with print_lock:
print(s, file=sys.stderr, flush=True)
async def log_event(event: Dict[str, Any]):
try:
async with event_log_lock:
append_jsonl(EVENT_LOG_PATH, event)
except Exception as e:
await log_status(f"[{ts()}] event_log write error path={EVENT_LOG_PATH!r} err={e!r}")
DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0"))
recent_payloads: Dict[bytes, float] = {}
@@ -416,6 +427,33 @@ async def main():
async def message_handler(msg: nats.aio.msg.Msg):
t_start = monotonic()
payload = msg.data
event = {
"ts": ts_iso(),
"event_type": "registration_attempt",
"subject": msg.subject,
"product": "-",
"mac": "-",
"mac_norm": None,
"fw": "-",
"host": None,
"iface_id": None,
"nb_lookup_result": "not_attempted",
"ignored": False,
"ignore_reason": None,
"action_next": None,
"action_last": None,
"action_state": None,
"decision": "received",
"decision_reason": None,
"publish_attempted": False,
"publish_result": "not_attempted",
"routing_key": None,
"task_name": None,
"delay_ms": None,
"netbox_ms": None,
"total_ms": None,
"nb_problems": None,
}
# strict payload dedupe
digest = hashlib.blake2b(payload, digest_size=16).digest()
@@ -438,8 +476,21 @@ async def main():
except Exception:
pass
event["product"] = product
event["mac"] = mac
event["fw"] = fw
mac_norm = normalize_mac(mac)
event["mac_norm"] = mac_norm
if mac_norm in IGNORE_MACS:
event["ignored"] = True
event["ignore_reason"] = "mac_ignore_list"
event["decision"] = "ignored"
event["decision_reason"] = "MAC is in local ignore list"
event["total_ms"] = round((monotonic() - t_start) * 1000, 1)
async with NB_PROBLEM_LOCK:
event["nb_problems"] = NB_PROBLEM_COUNTER
await log_event(event)
async with print_lock:
print(
f"[{ts()}] ignoring registration for mac={mac}",
@@ -459,8 +510,18 @@ async def main():
host, iface_id, dev_id, status_val, tag_slugs, action_next, action_last, action_next_timestamp, action_state, sot_ts = nb_lookup_device_by_mac(
mac=mac, log_status=log_status
)
event["iface_id"] = iface_id
event["host"] = host
event["action_next"] = action_next
event["action_last"] = action_last
event["action_state"] = action_state
if host:
event["nb_lookup_result"] = "resolved"
host_suffix = f" host={host}"
elif iface_id:
event["nb_lookup_result"] = "iface_only"
else:
event["nb_lookup_result"] = "not_found"
if iface_id and not host:
host_suffix += f" iface_id={iface_id}"
@@ -478,6 +539,14 @@ async def main():
action_state_str = ""
if action_state_str not in ("", "ready", "done"):
event["action_state"] = action_state_str
event["decision"] = "blocked_action_state"
event["decision_reason"] = "device is not ready because of action_state"
event["netbox_ms"] = round((time.perf_counter() - nb_start) * 1000, 1)
event["total_ms"] = round((monotonic() - t_start) * 1000, 1)
async with NB_PROBLEM_LOCK:
event["nb_problems"] = NB_PROBLEM_COUNTER
await log_event(event)
async with print_lock:
print(
f"[{ts()}] device is not ready because of action_state host={host} action_state={action_state_str}",
@@ -502,6 +571,8 @@ async def main():
action_next_str = None
if not has_action_next:
event["decision"] = "no_action"
event["decision_reason"] = "no action_next for host"
# 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)
@@ -522,6 +593,8 @@ async def main():
# Gate: if action_next is posture_analyzer and sot_ts is recent, skip sending task
skip_due_sot = False
event["action_next"] = action_next_str
event["action_last"] = action_last_str
if action_next_str == posture_analyzer:
try:
if isinstance(sot_ts, str):
@@ -531,6 +604,8 @@ async def main():
_age = now_epoch - int(_dt.timestamp())
if _age >= 0 and _age < sot_timeout:
skip_due_sot = True
event["decision"] = "cooldown_sot"
event["decision_reason"] = f"sot_ts is recent age_s={_age}"
async with print_lock:
print(
f"[{ts()}] skip action_next because sot_ts is recent host={host} task={action_next_str} age_s={_age} sot_ts={_st}",
@@ -558,6 +633,8 @@ async def main():
allow_repeat = True
if not allow_repeat:
event["decision"] = "cooldown"
event["decision_reason"] = "action_next cooldown active"
async with print_lock:
print(
f"[{ts()}] cooldown action_next for host={host} task={action_next_str}",
@@ -570,6 +647,10 @@ async def main():
target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish"
event["publish_attempted"] = True
event["publish_result"] = "attempted"
event["routing_key"] = RMQ_ROUTING_KEY
event["task_name"] = action_next_str
payload_obj = {
"inscope_device": host,
"task_name": action_next_str,
@@ -588,6 +669,7 @@ async def main():
if effective_delay_ms > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)}
event["delay_ms"] = int(effective_delay_ms)
# ---- SURGICAL FIX (bell must sound during payload_raw line) ----
async with print_lock:
@@ -605,6 +687,9 @@ async def main():
resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT)
if code != 200:
event["decision"] = "publish_failed"
event["decision_reason"] = f"rmq publish http={code}"
event["publish_result"] = "failed"
await log_status(f"[{ts()}] rmq: publish http={code} host={host}")
else:
routed = False
@@ -618,10 +703,16 @@ async def main():
publish_ok = True
if effective_delay_ms <= 0 and not routed:
publish_ok = False
event["decision"] = "publish_failed"
event["decision_reason"] = "rmq immediate publish routed=false"
event["publish_result"] = "failed"
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
# ----------------------
if publish_ok:
event["decision"] = "scheduled"
event["decision_reason"] = "publish sent to rabbitmq"
event["publish_result"] = "sent"
# On success: set action_last and action_next_timestamp and action_state
try:
base = NB_URL.rstrip("/")
@@ -648,6 +739,10 @@ async def main():
target_exchange = RMQ_EXCHANGE_DELAYED if effective_delay_ms > 0 else RMQ_EXCHANGE_WORK
rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{target_exchange}/publish"
event["publish_attempted"] = True
event["publish_result"] = "attempted"
event["routing_key"] = RMQ_ROUTING_KEY
event["task_name"] = action_next_str
payload_obj = {
"inscope_device": host,
"task_name": action_next_str,
@@ -666,6 +761,7 @@ async def main():
if effective_delay_ms > 0:
rmq_body["properties"]["headers"] = {"x-delay": int(effective_delay_ms)}
event["delay_ms"] = int(effective_delay_ms)
# ---- SURGICAL FIX (bell must sound during payload_raw line) ----
async with print_lock:
@@ -683,6 +779,9 @@ async def main():
resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT)
if code != 200:
event["decision"] = "publish_failed"
event["decision_reason"] = f"rmq publish http={code}"
event["publish_result"] = "failed"
await log_status(f"[{ts()}] rmq: publish http={code} host={host}")
else:
routed = False
@@ -695,10 +794,16 @@ async def main():
publish_ok = True
if effective_delay_ms <= 0 and not routed:
publish_ok = False
event["decision"] = "publish_failed"
event["decision_reason"] = "rmq immediate publish routed=false"
event["publish_result"] = "failed"
await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
# ----------------------
if publish_ok:
event["decision"] = "scheduled"
event["decision_reason"] = "publish sent to rabbitmq"
event["publish_result"] = "sent"
now_epoch = int(time.time())
try:
base = NB_URL.rstrip("/")
@@ -721,6 +826,9 @@ async def main():
bell_prefix = ""
except Exception as e:
event["decision"] = "error"
event["decision_reason"] = f"nb unexpected error {e!r}"
event["nb_lookup_result"] = "error"
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
@@ -729,6 +837,24 @@ async def main():
async with NB_PROBLEM_LOCK:
nb_problems_snapshot = NB_PROBLEM_COUNTER
event["netbox_ms"] = round(netbox_time_ms, 1) if product == "fox100" else None
event["total_ms"] = round(total_ms, 1)
event["nb_problems"] = nb_problems_snapshot
if event["decision"] == "received":
if product != "fox100":
event["decision"] = "unsupported_product"
event["decision_reason"] = "product is not handled by fox100 logic"
elif event["nb_lookup_result"] == "resolved":
event["decision"] = "lookup_only"
event["decision_reason"] = "registration processed without further action"
elif event["nb_lookup_result"] == "iface_only":
event["decision"] = "lookup_partial"
event["decision_reason"] = "mac resolved to interface only"
elif event["nb_lookup_result"] == "not_found":
event["decision"] = "lookup_failed"
event["decision_reason"] = "mac not found in netbox"
await log_event(event)
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}"