feat(agent): heartbeat job recording to sync_runs
This commit is contained in:
@@ -0,0 +1,34 @@
|
|||||||
|
"""Heartbeat: prove Firefly API + agentdb connectivity on a schedule."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
from agent.config import AgentConfig
|
||||||
|
from agent.db import connect
|
||||||
|
from agent.firefly import FireflyClient
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def run_heartbeat(cfg: AgentConfig, client: FireflyClient | None = None) -> str:
|
||||||
|
if client is None:
|
||||||
|
client = FireflyClient(cfg.firefly_url, cfg.firefly_token)
|
||||||
|
|
||||||
|
started_at = datetime.now(timezone.utc)
|
||||||
|
try:
|
||||||
|
info = client.about()
|
||||||
|
status, detail = "ok", f"firefly version {info.get('version', 'unknown')}"
|
||||||
|
except Exception as exc: # any failure is a recorded outcome, not a crash
|
||||||
|
status, detail = "error", f"{type(exc).__name__}: {exc}"
|
||||||
|
logger.warning("heartbeat failed: %s", detail)
|
||||||
|
|
||||||
|
finished_at = datetime.now(timezone.utc)
|
||||||
|
with connect(cfg) as conn:
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO sync_runs (connector, started_at, finished_at, status, detail)"
|
||||||
|
" VALUES (%s, %s, %s, %s, %s)",
|
||||||
|
("heartbeat", started_at, finished_at, status, detail),
|
||||||
|
)
|
||||||
|
return status
|
||||||
@@ -0,0 +1,48 @@
|
|||||||
|
from agent.db import connect, ensure_database, run_migrations
|
||||||
|
from agent.firefly import FireflyError
|
||||||
|
from agent.jobs.heartbeat import run_heartbeat
|
||||||
|
|
||||||
|
|
||||||
|
class FakeFirefly:
|
||||||
|
def __init__(self, version: str | None = None, error: Exception | None = None):
|
||||||
|
self._version = version
|
||||||
|
self._error = error
|
||||||
|
|
||||||
|
def about(self) -> dict:
|
||||||
|
if self._error:
|
||||||
|
raise self._error
|
||||||
|
return {"version": self._version}
|
||||||
|
|
||||||
|
|
||||||
|
def _last_run(cfg):
|
||||||
|
with connect(cfg) as conn:
|
||||||
|
return conn.execute(
|
||||||
|
"SELECT connector, status, detail, started_at, finished_at"
|
||||||
|
" FROM sync_runs ORDER BY id DESC LIMIT 1"
|
||||||
|
).fetchone()
|
||||||
|
|
||||||
|
|
||||||
|
def test_heartbeat_records_ok(cfg):
|
||||||
|
ensure_database(cfg)
|
||||||
|
run_migrations(cfg)
|
||||||
|
|
||||||
|
status = run_heartbeat(cfg, client=FakeFirefly(version="6.1.1"))
|
||||||
|
|
||||||
|
assert status == "ok"
|
||||||
|
connector, db_status, detail, started_at, finished_at = _last_run(cfg)
|
||||||
|
assert connector == "heartbeat"
|
||||||
|
assert db_status == "ok"
|
||||||
|
assert "6.1.1" in detail
|
||||||
|
assert finished_at >= started_at
|
||||||
|
|
||||||
|
|
||||||
|
def test_heartbeat_records_error_and_does_not_raise(cfg):
|
||||||
|
ensure_database(cfg)
|
||||||
|
run_migrations(cfg)
|
||||||
|
|
||||||
|
status = run_heartbeat(cfg, client=FakeFirefly(error=FireflyError("GET /api/v1/about -> 401")))
|
||||||
|
|
||||||
|
assert status == "error"
|
||||||
|
connector, db_status, detail, *_ = _last_run(cfg)
|
||||||
|
assert db_status == "error"
|
||||||
|
assert "401" in detail
|
||||||
Reference in New Issue
Block a user