health: background refresher thread (request path never blocks); wrapper escalates TERM→KILL
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2 changed files +38 −15
modified
infra/scripts/run-under-sshd.sh
+9 −2
@@ -12,6 +12,8 @@ KEY="$HOME/.ssh/id_ed25519_maclustr" | |||||
| 12 | REMOTE_PATH="/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:$HOME/.local/bin:/usr/bin:/bin:/usr/sbin:/sbin" | 12 | REMOTE_PATH="/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:$HOME/.local/bin:/usr/bin:/bin:/usr/sbin:/sbin" | ||
| 13 | NAME="${name:-findata-$$}" # PM2 exports the process name as $name | 13 | NAME="${name:-findata-$$}" # PM2 exports the process name as $name | ||
| 14 | PIDFILE="$WORKDIR/logs/sshd-$NAME.pid" | 14 | PIDFILE="$WORKDIR/logs/sshd-$NAME.pid" | ||
| 15 | +GRACE="${FINDATA_STOP_GRACE:-3600}" # seconds to wait after SIGTERM before SIGKILL (workers finish their job) | ||||
| 16 | +case "$NAME" in *api*) GRACE="${FINDATA_STOP_GRACE:-30}";; esac | ||||
| 15 | mkdir -p "$WORKDIR/logs" | 17 | mkdir -p "$WORKDIR/logs" | ||
| 16 | SSH=(ssh -T -q -o BatchMode=yes -o IdentitiesOnly=yes -i "$KEY" -o StrictHostKeyChecking=accept-new -o ServerAliveInterval=30 -o ServerAliveCountMax=3 localhost) | 18 | SSH=(ssh -T -q -o BatchMode=yes -o IdentitiesOnly=yes -i "$KEY" -o StrictHostKeyChecking=accept-new -o ServerAliveInterval=30 -o ServerAliveCountMax=3 localhost) | ||
| 17 | if [ -f "$PIDFILE" ]; then | 19 | if [ -f "$PIDFILE" ]; then | ||
@@ -19,7 +21,8 @@ if [ -f "$PIDFILE" ]; then | |||||
| 19 | if [ -n "$OLD" ] && kill -0 "$OLD" 2>/dev/null; then | 21 | if [ -n "$OLD" ] && kill -0 "$OLD" 2>/dev/null; then | ||
| 20 | echo "run-under-sshd: terminating leftover remote process $OLD for $NAME" >&2 | 22 | echo "run-under-sshd: terminating leftover remote process $OLD for $NAME" >&2 | ||
| 21 | kill -TERM "$OLD" 2>/dev/null | 23 | kill -TERM "$OLD" 2>/dev/null | ||
| 22 | − for _ in $(seq 1 3600); do kill -0 "$OLD" 2>/dev/null || break; sleep 1; done | 24 | + for _ in $(seq 1 "$GRACE"); do kill -0 "$OLD" 2>/dev/null || break; sleep 1; done | ||
| 25 | + kill -0 "$OLD" 2>/dev/null && { echo "run-under-sshd: SIGKILL leftover $OLD" >&2; kill -KILL "$OLD" 2>/dev/null; sleep 1; } | ||||
| 23 | fi | 26 | fi | ||
| 24 | fi | 27 | fi | ||
| 25 | CMD=$(printf '%q ' "$@") | 28 | CMD=$(printf '%q ' "$@") | ||
@@ -27,7 +30,11 @@ CMD=$(printf '%q ' "$@") | |||||
| 27 | SSHPID=$! | 30 | SSHPID=$! | ||
| 28 | stop() { | 31 | stop() { | ||
| 29 | REMOTE=$(cat "$PIDFILE" 2>/dev/null) | 32 | REMOTE=$(cat "$PIDFILE" 2>/dev/null) | ||
| 30 | − [ -n "$REMOTE" ] && kill -TERM "$REMOTE" 2>/dev/null | 33 | + if [ -n "$REMOTE" ]; then | ||
| 34 | + kill -TERM "$REMOTE" 2>/dev/null | ||||
| 35 | + for _ in $(seq 1 "$GRACE"); do kill -0 "$REMOTE" 2>/dev/null || break; sleep 1; done | ||||
| 36 | + kill -0 "$REMOTE" 2>/dev/null && kill -KILL "$REMOTE" 2>/dev/null | ||||
| 37 | + fi | ||||
| 31 | wait "$SSHPID" 2>/dev/null | 38 | wait "$SSHPID" 2>/dev/null | ||
| 32 | rm -f "$PIDFILE" | 39 | rm -f "$PIDFILE" | ||
| 33 | exit 0 | 40 | exit 0 | ||
modified
services/api/src/findata_api/health.py
+29 −13
@@ -11,22 +11,38 @@ from findata_ingest.lake import Lake | |||||
| 11 | from findata_api import metrics as m | 11 | from findata_api import metrics as m | ||
| 12 | 12 | ||||
| 13 | _cache: dict[str, Any] = {"at": 0.0, "value": None} | 13 | _cache: dict[str, Any] = {"at": 0.0, "value": None} | ||
| 14 | −_lock = __import__("threading").Lock() | 14 | +_refresher_started = False | ||
| 15 | 15 | ||||
| 16 | 16 | ||||
| 17 | −def cached_evaluate(max_age_s: float = 60.0) -> dict[str, Any]: | 17 | +def _refresh_loop(period_s: float) -> None: | ||
| 18 | − """Health evaluation is served from a 60 s cache so that probes (mld heal, agentd, Prometheus) stay cheap | ||||
| 19 | − even while hundreds of thousands of files are being published.""" | ||||
| 20 | import time | 18 | import time | ||
| 21 | − now = time.monotonic() | 19 | + while True: | ||
| 22 | − if _cache["value"] is not None and now - _cache["at"] < max_age_s: | 20 | + try: | ||
| 23 | − return _cache["value"] | 21 | + v = evaluate() | ||
| 24 | − with _lock: | 22 | + _cache.update(at=time.monotonic(), value=v) | ||
| 25 | − if _cache["value"] is not None and time.monotonic() - _cache["at"] < max_age_s: | 23 | + except Exception as e: | ||
| 26 | − return _cache["value"] | 24 | + _cache.update(at=time.monotonic(), value={"status": "degraded", "checked_at": datetime.now(UTC).isoformat(), | ||
| 27 | − v = evaluate() | 25 | + "alerts": [{"severity": "error", "code": "health_eval_failed", "message": str(e)[:200]}], | ||
| 28 | − _cache.update(at=time.monotonic(), value=v) | 26 | + "components": {}}) | ||
| 29 | − return v | 27 | + time.sleep(period_s) | ||
| 28 | + | ||||
| 29 | + | ||||
| 30 | +def start_refresher(period_s: float = 60.0) -> None: | ||||
| 31 | + """Health is computed by a background thread; request handlers only read the cache (never block on DB/disk).""" | ||||
| 32 | + global _refresher_started | ||||
| 33 | + if _refresher_started: | ||||
| 34 | + return | ||||
| 35 | + import threading | ||||
| 36 | + _refresher_started = True | ||||
| 37 | + threading.Thread(target=_refresh_loop, args=(period_s,), name="health-refresher", daemon=True).start() | ||||
| 38 | + | ||||
| 39 | + | ||||
| 40 | +def cached_evaluate(max_age_s: float = 60.0) -> dict[str, Any]: | ||||
| 41 | + start_refresher() | ||||
| 42 | + v = _cache["value"] | ||||
| 43 | + if v is None: | ||||
| 44 | + return {"status": "starting", "checked_at": datetime.now(UTC).isoformat(), "alerts": [], "components": {}, "datasets": []} | ||||
| 45 | + return v | ||||
| 30 | 46 | ||||
| 31 | 47 | ||||
| 32 | def lake_usage_bytes(conn) -> dict[str, int]: | 48 | def lake_usage_bytes(conn) -> dict[str, int]: | ||
| 33 | 49 | ||||