mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-15 14:22:43 +00:00
fix(dashboard): cache gateway PID status probes
(cherry picked from commit a2bbe564ad)
This commit is contained in:
parent
49fa04a235
commit
7d0ddbb2ff
4 changed files with 182 additions and 15 deletions
|
|
@ -18,6 +18,8 @@ import shlex
|
|||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from hermes_constants import get_hermes_home, _get_platform_default_hermes_home
|
||||
|
|
@ -40,6 +42,9 @@ _gateway_lock_handle = None
|
|||
# past the JSON payload so runtime status / PID readers can still read the file
|
||||
# while another process holds the mutual-exclusion lock.
|
||||
_WINDOWS_LOCK_OFFSET = 1024 * 1024
|
||||
_GATEWAY_RUNNING_PID_CACHE_TTL_SECONDS = 1.0
|
||||
_gateway_running_pid_cache_lock = threading.Lock()
|
||||
_gateway_running_pid_cache: dict[tuple[str, bool, bool], tuple[float, tuple[Any, ...], Optional[int]]] = {}
|
||||
|
||||
|
||||
def _get_process_hermes_home() -> Path:
|
||||
|
|
@ -496,6 +501,33 @@ def _pid_from_record(record: Optional[dict[str, Any]]) -> Optional[int]:
|
|||
return None
|
||||
|
||||
|
||||
def _clear_running_pid_cache() -> None:
|
||||
with _gateway_running_pid_cache_lock:
|
||||
_gateway_running_pid_cache.clear()
|
||||
|
||||
|
||||
def _file_cache_signature(path: Path) -> tuple[bool, Optional[int], Optional[int]]:
|
||||
try:
|
||||
st = path.stat()
|
||||
except OSError:
|
||||
return (False, None, None)
|
||||
return (True, st.st_mtime_ns, st.st_size)
|
||||
|
||||
|
||||
def _running_pid_cache_signature(
|
||||
pid_path: Path,
|
||||
*,
|
||||
include_runtime_status: bool,
|
||||
) -> tuple[Any, ...]:
|
||||
parts: list[Any] = [
|
||||
_file_cache_signature(pid_path),
|
||||
_file_cache_signature(_get_gateway_lock_path(pid_path)),
|
||||
]
|
||||
if include_runtime_status:
|
||||
parts.append(_file_cache_signature(_get_runtime_status_path()))
|
||||
return tuple(parts)
|
||||
|
||||
|
||||
def _cleanup_invalid_pid_path(pid_path: Path, *, cleanup_stale: bool) -> None:
|
||||
"""Delete a stale gateway PID file (and its sibling lock metadata).
|
||||
|
||||
|
|
@ -508,6 +540,7 @@ def _cleanup_invalid_pid_path(pid_path: Path, *, cleanup_stale: bool) -> None:
|
|||
"""
|
||||
if not cleanup_stale:
|
||||
return
|
||||
_clear_running_pid_cache()
|
||||
try:
|
||||
pid_path.unlink(missing_ok=True)
|
||||
except Exception:
|
||||
|
|
@ -693,6 +726,7 @@ def acquire_gateway_runtime_lock() -> bool:
|
|||
return False
|
||||
_write_gateway_lock_record(handle)
|
||||
_gateway_lock_handle = handle
|
||||
_clear_running_pid_cache()
|
||||
return True
|
||||
|
||||
|
||||
|
|
@ -708,6 +742,7 @@ def release_gateway_runtime_lock() -> None:
|
|||
handle.close()
|
||||
except OSError:
|
||||
pass
|
||||
_clear_running_pid_cache()
|
||||
|
||||
|
||||
def is_gateway_runtime_lock_active(lock_path: Optional[Path] = None) -> bool:
|
||||
|
|
@ -750,6 +785,7 @@ def write_pid_file() -> None:
|
|||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
f.write(record)
|
||||
_clear_running_pid_cache()
|
||||
except Exception:
|
||||
try:
|
||||
path.unlink(missing_ok=True)
|
||||
|
|
@ -942,6 +978,7 @@ def remove_pid_file() -> None:
|
|||
# PID file belongs to a different process — leave it alone.
|
||||
return
|
||||
path.unlink(missing_ok=True)
|
||||
_clear_running_pid_cache()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
|
@ -1447,6 +1484,54 @@ def get_running_pid(
|
|||
return None
|
||||
|
||||
|
||||
def get_running_pid_cached(
|
||||
pid_path: Optional[Path] = None,
|
||||
*,
|
||||
cleanup_stale: bool = True,
|
||||
ttl_seconds: float = _GATEWAY_RUNNING_PID_CACHE_TTL_SECONDS,
|
||||
) -> Optional[int]:
|
||||
"""Cached read-side wrapper for dashboard/status polling.
|
||||
|
||||
``get_running_pid()`` probes the runtime lock by briefly opening and locking
|
||||
``gateway.lock``. That is the right authoritative check for control paths,
|
||||
but high-frequency read-only HTTP polling can call it hundreds of times per
|
||||
minute. Cache for a short window and invalidate on PID/lock/runtime-status
|
||||
file changes so status endpoints do not churn file descriptors while still
|
||||
noticing gateway start/stop transitions quickly.
|
||||
"""
|
||||
if ttl_seconds <= 0:
|
||||
return get_running_pid(pid_path, cleanup_stale=cleanup_stale)
|
||||
|
||||
resolved_pid_path = pid_path or _get_pid_path()
|
||||
include_runtime_status = pid_path is None
|
||||
signature = _running_pid_cache_signature(
|
||||
resolved_pid_path,
|
||||
include_runtime_status=include_runtime_status,
|
||||
)
|
||||
key = (str(resolved_pid_path), bool(cleanup_stale), include_runtime_status)
|
||||
now = time.monotonic()
|
||||
|
||||
with _gateway_running_pid_cache_lock:
|
||||
cached = _gateway_running_pid_cache.get(key)
|
||||
if cached is not None:
|
||||
cached_at, cached_signature, cached_pid = cached
|
||||
if now - cached_at <= ttl_seconds and cached_signature == signature:
|
||||
return cached_pid
|
||||
|
||||
pid = get_running_pid(pid_path, cleanup_stale=cleanup_stale)
|
||||
refreshed_signature = _running_pid_cache_signature(
|
||||
resolved_pid_path,
|
||||
include_runtime_status=include_runtime_status,
|
||||
)
|
||||
with _gateway_running_pid_cache_lock:
|
||||
_gateway_running_pid_cache[key] = (
|
||||
time.monotonic(),
|
||||
refreshed_signature,
|
||||
pid,
|
||||
)
|
||||
return pid
|
||||
|
||||
|
||||
def is_gateway_running(
|
||||
pid_path: Optional[Path] = None,
|
||||
*,
|
||||
|
|
|
|||
|
|
@ -78,6 +78,7 @@ from hermes_cli.config import (
|
|||
from gateway.status import (
|
||||
derive_gateway_busy,
|
||||
derive_gateway_drainable,
|
||||
get_running_pid_cached,
|
||||
get_running_pid,
|
||||
get_runtime_status_running_pid,
|
||||
parse_active_agents,
|
||||
|
|
@ -2456,7 +2457,7 @@ async def get_status(profile: Optional[str] = None):
|
|||
# Try local PID check first (same-host). If that fails and a remote
|
||||
# GATEWAY_HEALTH_URL is configured, probe the gateway over HTTP so the
|
||||
# dashboard works when the gateway runs in a separate container.
|
||||
gateway_pid = get_running_pid()
|
||||
gateway_pid = get_running_pid_cached()
|
||||
gateway_running = gateway_pid is not None
|
||||
remote_health_body: dict | None = None
|
||||
|
||||
|
|
|
|||
|
|
@ -223,6 +223,71 @@ class TestGatewayPidState:
|
|||
|
||||
assert status.get_running_pid() == os.getpid()
|
||||
|
||||
def test_get_running_pid_cached_reuses_runtime_lock_probe(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
status._clear_running_pid_cache()
|
||||
|
||||
pid_path = tmp_path / "gateway.pid"
|
||||
record = {
|
||||
"pid": os.getpid(),
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["python", "-m", "hermes_cli.main", "gateway"],
|
||||
"start_time": 123,
|
||||
}
|
||||
pid_path.write_text(json.dumps(record))
|
||||
(tmp_path / "gateway.lock").write_text(json.dumps(record))
|
||||
|
||||
calls = {"lock_active": 0}
|
||||
|
||||
def _lock_active(lock_path=None):
|
||||
calls["lock_active"] += 1
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(status, "is_gateway_runtime_lock_active", _lock_active)
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda pid: True)
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda pid: 123)
|
||||
monkeypatch.setattr(status, "_read_process_cmdline", lambda pid: None)
|
||||
|
||||
assert status.get_running_pid_cached(ttl_seconds=60) == os.getpid()
|
||||
assert status.get_running_pid_cached(ttl_seconds=60) == os.getpid()
|
||||
assert calls["lock_active"] == 1
|
||||
|
||||
def test_get_running_pid_cached_invalidates_when_pid_file_changes(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
status._clear_running_pid_cache()
|
||||
|
||||
pid_path = tmp_path / "gateway.pid"
|
||||
|
||||
def _write_record(pid: int, start_time: int) -> None:
|
||||
record = {
|
||||
"pid": pid,
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["python", "-m", "hermes_cli.main", "gateway"],
|
||||
"start_time": start_time,
|
||||
}
|
||||
pid_path.write_text(json.dumps(record))
|
||||
(tmp_path / "gateway.lock").write_text(json.dumps(record))
|
||||
|
||||
_write_record(111, 123)
|
||||
|
||||
calls = {"lock_active": 0}
|
||||
|
||||
def _lock_active(lock_path=None):
|
||||
calls["lock_active"] += 1
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(status, "is_gateway_runtime_lock_active", _lock_active)
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda pid: True)
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda pid: 123 if pid == 111 else 456)
|
||||
monkeypatch.setattr(status, "_read_process_cmdline", lambda pid: None)
|
||||
|
||||
assert status.get_running_pid_cached(ttl_seconds=60) == 111
|
||||
|
||||
_write_record(2222, 456)
|
||||
|
||||
assert status.get_running_pid_cached(ttl_seconds=60) == 2222
|
||||
assert calls["lock_active"] == 2
|
||||
|
||||
def test_get_running_pid_cleans_stale_metadata_from_dead_foreign_pid(self, tmp_path, monkeypatch):
|
||||
"""Stale PID file from a *different* PID (crashed process) must still be cleaned.
|
||||
|
||||
|
|
|
|||
|
|
@ -305,6 +305,22 @@ class TestWebServerEndpoints:
|
|||
assert resp.status_code == 200
|
||||
assert resp.json()["active_sessions"] == 0
|
||||
|
||||
def test_get_status_uses_cached_gateway_pid_probe(self, monkeypatch):
|
||||
import hermes_cli.web_server as web_server
|
||||
|
||||
calls = {"get_running_pid_cached": 0}
|
||||
|
||||
def _cached_pid():
|
||||
calls["get_running_pid_cached"] += 1
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(web_server, "get_running_pid_cached", _cached_pid)
|
||||
|
||||
resp = self.client.get("/api/status")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert calls["get_running_pid_cached"] == 1
|
||||
|
||||
def test_gateway_drain_begin_writes_marker(self):
|
||||
from gateway import drain_control
|
||||
|
||||
|
|
@ -1617,7 +1633,7 @@ class TestWebServerEndpoints:
|
|||
def get_connected_platforms(self):
|
||||
return [_Platform("telegram")]
|
||||
|
||||
monkeypatch.setattr(web_server, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(web_server, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(
|
||||
web_server,
|
||||
"read_runtime_status",
|
||||
|
|
@ -1649,7 +1665,7 @@ class TestWebServerEndpoints:
|
|||
def get_connected_platforms(self):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(web_server, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(web_server, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(
|
||||
web_server,
|
||||
"read_runtime_status",
|
||||
|
|
@ -5057,7 +5073,7 @@ class TestStatusRemoteGateway:
|
|||
"""When local PID check fails and remote probe succeeds, gateway shows running."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
||||
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
||||
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
|
||||
|
|
@ -5079,7 +5095,7 @@ class TestStatusRemoteGateway:
|
|||
"""When local PID check succeeds, the remote probe is never called."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "running",
|
||||
"platforms": {},
|
||||
|
|
@ -5102,7 +5118,7 @@ class TestStatusRemoteGateway:
|
|||
"""When GATEWAY_HEALTH_URL is unset, no probe is attempted."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
||||
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
|
||||
|
||||
|
|
@ -5116,7 +5132,7 @@ class TestStatusRemoteGateway:
|
|||
"""Remote gateway running but PID not in response — pid should be None."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
||||
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
||||
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
|
||||
|
|
@ -5155,7 +5171,7 @@ class TestGatewayBusyReadout:
|
|||
"""gateway_busy is True iff running AND active_agents > 0."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "running",
|
||||
"platforms": {},
|
||||
|
|
@ -5173,7 +5189,7 @@ class TestGatewayBusyReadout:
|
|||
"""A running gateway with zero in-flight turns is drainable, not busy."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "running",
|
||||
"platforms": {},
|
||||
|
|
@ -5191,7 +5207,7 @@ class TestGatewayBusyReadout:
|
|||
gate dominates."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "draining",
|
||||
"platforms": {},
|
||||
|
|
@ -5207,7 +5223,7 @@ class TestGatewayBusyReadout:
|
|||
active_agents 0 — never a spurious busy that would wedge NAS."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
||||
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
|
||||
|
||||
|
|
@ -5223,9 +5239,9 @@ class TestGatewayBusyReadout:
|
|||
wins over the file."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: None)
|
||||
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
|
||||
# File says running with active turns, but get_running_pid()==None and
|
||||
# File says running with active turns, but get_running_pid_cached()==None and
|
||||
# get_runtime_status_running_pid finds no live PID → gateway_running False.
|
||||
monkeypatch.setattr(ws, "get_runtime_status_running_pid", lambda *_a, **_k: None)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
|
|
@ -5244,7 +5260,7 @@ class TestGatewayBusyReadout:
|
|||
float so NAS can size its poll deadline without out-of-band knowledge."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "running",
|
||||
"platforms": {},
|
||||
|
|
@ -5262,7 +5278,7 @@ class TestGatewayBusyReadout:
|
|||
produce a spurious busy — it degrades to 0/not-busy."""
|
||||
import hermes_cli.web_server as ws
|
||||
|
||||
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "get_running_pid_cached", lambda: 1234)
|
||||
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
||||
"gateway_state": "running",
|
||||
"platforms": {},
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue