127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""
|
|
Node probe — query a remote fester-agent for live state.
|
|
|
|
Each cluster node is expected to run an HTTP agent on port 8787 (the
|
|
convention established by backend/nodes_cli.py) that returns JSON like:
|
|
|
|
{
|
|
"state": "online",
|
|
"load": "0.42 0.51 0.48 2/128 1234",
|
|
"temp": 62,
|
|
"jobs": 4,
|
|
"max_jobs": 24,
|
|
"memory": {"used": 8.2, "total": 32.0, "unit": "GB"},
|
|
"labels": {"arch": "x86_64", "role": "compile"}
|
|
}
|
|
|
|
If the agent is unreachable, the probe returns None and the caller can
|
|
fall back to synthetic drift.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
try:
|
|
import aiohttp
|
|
_HAVE_AIOHTTP = True
|
|
except ImportError:
|
|
_HAVE_AIOHTTP = False
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
AGENT_PORT = 8787
|
|
DEFAULT_TIMEOUT = 1.5 # seconds
|
|
|
|
|
|
def probe_sync(host: str, port: int = AGENT_PORT, timeout: float = DEFAULT_TIMEOUT) -> Optional[Dict[str, Any]]:
|
|
"""Synchronous probe — for use outside the event loop."""
|
|
if not host:
|
|
return None
|
|
url = f"http://{host}:{port}/status"
|
|
try:
|
|
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
import json
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError):
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def probe_async(host: str, port: int = AGENT_PORT, timeout: float = DEFAULT_TIMEOUT) -> Optional[Dict[str, Any]]:
|
|
"""Async probe — preferred when running inside FastAPI / asyncio."""
|
|
if not host:
|
|
return None
|
|
url = f"http://{host}:{port}/status"
|
|
if _HAVE_AIOHTTP:
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session:
|
|
async with session.get(url) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
return await resp.json()
|
|
except (asyncio.TimeoutError, aiohttp.ClientError, OSError):
|
|
return None
|
|
except Exception:
|
|
return None
|
|
else:
|
|
# Fallback: run sync probe in a thread
|
|
return await asyncio.to_thread(probe_sync, host, port, timeout)
|
|
|
|
|
|
def normalize(agent_data: Optional[Dict[str, Any]], node_name: str) -> Dict[str, Any]:
|
|
"""Convert an agent's response into a normalized NodeState shape.
|
|
|
|
If agent_data is None (probe failed), returns a minimal offline shape."""
|
|
if not agent_data:
|
|
return {
|
|
"name": node_name,
|
|
"state": "offline",
|
|
"cpu_load": 0.0,
|
|
"memory_load": 0.0,
|
|
"temp": 0.0,
|
|
"instability": 1.0, # mark unstable since we can't reach it
|
|
"active_jobs": 0,
|
|
"last_seen": 0.0,
|
|
}
|
|
|
|
# Parse load average string like "0.42 0.51 0.48 2/128 1234"
|
|
load_str = agent_data.get("load", "0 0 0")
|
|
try:
|
|
cpu_load = float(str(load_str).split()[0]) * 100 # 1.0 = 100%
|
|
except (ValueError, IndexError):
|
|
cpu_load = 0.0
|
|
|
|
# Memory: agent reports {used, total, unit} in GB
|
|
mem = agent_data.get("memory", {})
|
|
if isinstance(mem, dict) and mem.get("total"):
|
|
memory_load = (float(mem.get("used", 0)) / float(mem["total"])) * 100
|
|
else:
|
|
memory_load = 0.0
|
|
|
|
return {
|
|
"name": node_name,
|
|
"state": agent_data.get("state", "online"),
|
|
"cpu_load": cpu_load,
|
|
"memory_load": memory_load,
|
|
"temp": float(agent_data.get("temp", 0)),
|
|
"instability": float(agent_data.get("instability", 0.0)),
|
|
"active_jobs": int(agent_data.get("jobs", 0)),
|
|
"max_jobs": int(agent_data.get("max_jobs", 0)),
|
|
"labels": agent_data.get("labels", {}),
|
|
"last_seen": time.time(),
|
|
}
|
|
|
|
|
|
async def probe_node(name: str, host: str, port: int = AGENT_PORT) -> Dict[str, Any]:
|
|
"""Probe one node and return normalized state.
|
|
Always returns a dict (offline if probe failed)."""
|
|
raw = await probe_async(host, port)
|
|
return normalize(raw, name)
|