creator/monlet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

store only status-change events; UI client-side polling refactor

Browse Source
This commit is contained in:
Stanislav Rossovskii
2026-05-27 14:58:46 +04:00
parent 71f0035b0b
commit d5f312f200
48 changed files with 2247 additions and 753 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
server/alembic/versions/0001_baseline.py
View File

@@ -43,6 +43,7 @@ def upgrade() -> None:
last_observed_at TIMESTAMPTZ NOT NULL,
last_event_id UUID NOT NULL,
incident_key TEXT NOT NULL,
summary TEXT,
PRIMARY KEY (agent_id, check_id)
);
"""

1
server/monlet_server/api/checks.py
View File

@@ -42,6 +42,7 @@ async def list_checks(
last_observed_at=to_utc(r.last_observed_at),
exit_code=r.exit_code,
incident_key=r.incident_key,
summary=r.summary,
)
for r in rows
]

1
server/monlet_server/models.py
View File

@@ -71,6 +71,7 @@ class Check(Base):
last_observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_event_id: Mapped[UUID] = mapped_column(_uuid(), nullable=False)
incident_key: Mapped[str] = mapped_column(Text, nullable=False)
summary: Mapped[str | None] = mapped_column(Text, nullable=True)
__table_args__ = (
PrimaryKeyConstraint("agent_id", "check_id", name="checks_pkey"),

1
server/monlet_server/schemas.py
View File

@@ -152,6 +152,7 @@ class CheckState(BaseModel):
last_observed_at: datetime
exit_code: int
incident_key: str
summary: str | None = None
class Incident(BaseModel):

169
server/monlet_server/services/detector.py
View File

@@ -1,7 +1,9 @@
import asyncio
from datetime import UTC, datetime, timedelta
from uuid import uuid4
from sqlalchemy import delete, func, select, update
from sqlalchemy import delete, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import async_sessionmaker
from .. import metrics
@@ -12,6 +14,138 @@ from ..settings import get_settings
log = get_logger("monlet.detector")
OUTBOX_TERMINAL_STATES = ("sent", "failed", "discarded")
LIVENESS_CHECK_ID = "agent_liveness"
def _liveness_key(agent_id: str) -> str:
return f"agent:{agent_id}:liveness"
def _liveness_severity(status: str) -> str:
return "critical" if status == "dead" else "warning"
async def _enqueue_liveness_outbox(
session,
incident: Incident,
event_type: str,
agent: Agent,
status: str,
now: datetime,
summary: str,
severity: str,
) -> None:
notifiers = get_settings().enabled_notifiers
if not notifiers:
return
for name in notifiers:
session.add(
NotificationOutbox(
id=uuid4(),
incident_id=incident.id,
notifier=name,
event_type=event_type,
state="pending",
attempts=0,
next_attempt_at=now,
payload={
"incident_id": str(incident.id),
"agent_id": agent.agent_id,
"check_id": LIVENESS_CHECK_ID,
"status": status,
"severity": severity,
"incident_key": incident.incident_key,
"observed_at": now.isoformat(),
"opened_at": incident.opened_at.isoformat() if incident.opened_at else None,
"resolved_at": incident.resolved_at.isoformat()
if incident.resolved_at
else None,
"summary": summary,
"output": summary,
"event_type": event_type,
},
)
)
async def _open_liveness_incident(session, agent: Agent, status: str, now: datetime) -> None:
severity = _liveness_severity(status)
summary = f"agent is {status}"
incident_key = _liveness_key(agent.agent_id)
new_id = uuid4()
stmt = (
pg_insert(Incident)
.values(
id=new_id,
incident_key=incident_key,
agent_id=agent.agent_id,
check_id=LIVENESS_CHECK_ID,
state="open",
severity=severity,
opened_at=now,
summary=summary,
last_event_id=uuid4(),
)
.on_conflict_do_nothing(
index_elements=[Incident.incident_key],
index_where=text("state = 'open'"),
)
.returning(Incident.id)
)
inserted_id = (await session.execute(stmt)).scalar_one_or_none()
if inserted_id is not None:
await session.flush()
inc = (
await session.execute(select(Incident).where(Incident.id == inserted_id))
).scalar_one()
metrics.incidents_opened_total.inc()
await _enqueue_liveness_outbox(
session, inc, "firing", agent, status, now, summary=summary, severity=severity
)
return
inc = (
await session.execute(
select(Incident).where(
Incident.incident_key == incident_key,
Incident.state == "open",
)
)
).scalar_one()
if status == "dead" and inc.severity != "critical":
inc.severity = "critical"
inc.summary = summary
inc.last_event_id = uuid4()
await _enqueue_liveness_outbox(
session, inc, "firing", agent, status, now, summary=summary, severity="critical"
)
async def _resolve_liveness_incident(session, agent: Agent, now: datetime) -> None:
inc = (
await session.execute(
select(Incident).where(
Incident.incident_key == _liveness_key(agent.agent_id),
Incident.state == "open",
)
)
).scalar_one_or_none()
if inc is None:
return
inc.state = "resolved"
inc.resolved_at = now
inc.last_event_id = uuid4()
metrics.incidents_resolved_total.inc()
await _enqueue_liveness_outbox(
session,
inc,
"resolved",
agent,
"ok",
now,
summary="agent is alive",
severity="ok",
)
async def _prune_history(session) -> None:
@@ -44,25 +178,20 @@ async def _tick(sm: async_sessionmaker) -> None:
stale_cut = now - timedelta(seconds=s.stale_after_sec)
dead_cut = now - timedelta(seconds=s.dead_after_sec)
async with sm() as session:
await session.execute(
update(Agent)
.where(Agent.last_seen_at <= dead_cut)
.where(Agent.status != "dead")
.values(status="dead")
)
await session.execute(
update(Agent)
.where(Agent.last_seen_at <= stale_cut)
.where(Agent.last_seen_at > dead_cut)
.where(Agent.status != "stale")
.values(status="stale")
)
await session.execute(
update(Agent)
.where(Agent.last_seen_at > stale_cut)
.where(Agent.status != "alive")
.values(status="alive")
)
agents = (await session.execute(select(Agent))).scalars().all()
for agent in agents:
if agent.last_seen_at <= dead_cut:
next_status = "dead"
elif agent.last_seen_at <= stale_cut:
next_status = "stale"
else:
next_status = "alive"
if agent.status != next_status:
agent.status = next_status
if next_status in ("stale", "dead"):
await _open_liveness_incident(session, agent, next_status, now)
else:
await _resolve_liveness_incident(session, agent, now)
await _prune_history(session)
await session.commit()

102
server/monlet_server/services/ingestion.py
View File

@@ -155,59 +155,79 @@ async def ingest_event(
agent_id: str,
ev: CheckResultEvent,
) -> bool:
"""Return True if accepted, False if deduplicated."""
"""Return True if a new change row was written, False if not (duplicate/unchanged/late)."""
observed = to_utc(ev.observed_at)
output = redact(ev.output)
check_summary = (output or "")[:200] if output else None
incident_key = _incident_key(agent_id, ev)
stmt = pg_insert(Event).values(
event_id=UUID(ev.event_id),
agent_id=agent_id,
check_id=ev.check_id,
observed_at=observed,
status=ev.status,
exit_code=ev.exit_code,
duration_ms=ev.duration_ms,
output=output,
output_truncated=ev.output_truncated,
notifications_enabled=ev.notifications_enabled,
incident_key=incident_key,
)
stmt = stmt.on_conflict_do_nothing(index_elements=[Event.event_id]).returning(Event.event_id)
result = await session.execute(stmt)
inserted = result.scalar_one_or_none()
if inserted is None:
metrics.events_total.labels(result="deduplicated").inc()
return False
metrics.events_total.labels(result="accepted").inc()
chk_res = await session.execute(
select(Check).where(Check.agent_id == agent_id, Check.check_id == ev.check_id)
)
cur = chk_res.scalar_one_or_none()
is_late = cur is not None and cur.last_observed_at > observed
if is_late:
return True
changed = cur is None or cur.status != ev.status or cur.exit_code != ev.exit_code
if cur is None:
session.add(
Check(
agent_id=agent_id,
check_id=ev.check_id,
status=ev.status,
exit_code=ev.exit_code,
last_observed_at=observed,
last_event_id=UUID(ev.event_id),
incident_key=incident_key,
)
if is_late:
metrics.events_total.labels(result="late").inc()
return False
if changed:
stmt = pg_insert(Event).values(
event_id=UUID(ev.event_id),
agent_id=agent_id,
check_id=ev.check_id,
observed_at=observed,
status=ev.status,
exit_code=ev.exit_code,
duration_ms=ev.duration_ms,
output=output,
output_truncated=ev.output_truncated,
notifications_enabled=ev.notifications_enabled,
incident_key=incident_key,
)
stmt = stmt.on_conflict_do_nothing(index_elements=[Event.event_id]).returning(
Event.event_id
)
result = await session.execute(stmt)
inserted = result.scalar_one_or_none()
if inserted is None:
metrics.events_total.labels(result="deduplicated").inc()
return False
metrics.events_total.labels(result="accepted").inc()
else:
cur.status = ev.status
cur.exit_code = ev.exit_code
cur.last_observed_at = observed
cur.last_event_id = UUID(ev.event_id)
cur.incident_key = incident_key
metrics.events_total.labels(result="unchanged").inc()
# Upsert Check to be safe under concurrent first-write race for the same (agent_id, check_id).
# When changed=False keep the existing last_event_id and incident_key — incident_key changes
# are only meaningful at transitions and would otherwise desync from any open Incident.
check_values = {
"agent_id": agent_id,
"check_id": ev.check_id,
"status": ev.status,
"exit_code": ev.exit_code,
"last_observed_at": observed,
"last_event_id": UUID(ev.event_id),
"incident_key": incident_key,
"summary": check_summary,
}
update_set = {
"status": ev.status,
"exit_code": ev.exit_code,
"last_observed_at": observed,
"summary": check_summary,
}
if changed:
update_set["last_event_id"] = UUID(ev.event_id)
update_set["incident_key"] = incident_key
chk_stmt = pg_insert(Check).values(**check_values)
chk_stmt = chk_stmt.on_conflict_do_update(
index_elements=[Check.agent_id, Check.check_id], set_=update_set
)
await session.execute(chk_stmt)
if not changed:
return False
transition, inc = await _apply_incident(session, agent_id, ev, incident_key)
if transition and inc is not None and ev.notifications_enabled:

2
server/monlet_server/settings.py
View File

@@ -20,7 +20,7 @@ class Settings(BaseSettings):
port: int = 8000
enable_detector: bool = True
enable_notifier_worker: bool = True
events_retention_max_rows: int = 100_000
events_retention_max_rows: int = 20_000
outbox_retention_max_rows: int = 50_000
notifier_tick_sec: int = 5
notifier_batch_size: int = 20

35
server/tests/test_detector.py
View File

@@ -6,7 +6,7 @@ from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import async_sessionmaker
from monlet_server.models import Agent, Event, Incident, NotificationOutbox
from monlet_server.services.detector import _tick
from monlet_server.services.detector import LIVENESS_CHECK_ID, _tick
from monlet_server.settings import reset_settings_cache
from ._helpers import make_event, make_heartbeat
@@ -39,6 +39,7 @@ async def test_detector_transitions(app_client, auth_headers, engine, session):
sm = async_sessionmaker(engine, expire_on_commit=False)
await _tick(sm)
session.expire_all()
statuses = {
r.agent_id: r.status for r in (await session.execute(select(Agent))).scalars().all()
@@ -47,6 +48,38 @@ async def test_detector_transitions(app_client, auth_headers, engine, session):
assert statuses["agent-stale"] == "stale"
assert statuses["agent-dead"] == "dead"
incidents = (
(await session.execute(select(Incident).where(Incident.check_id == LIVENESS_CHECK_ID)))
.scalars()
.all()
)
by_agent = {i.agent_id: i for i in incidents}
assert "agent-alive" not in by_agent
assert by_agent["agent-stale"].state == "open"
assert by_agent["agent-stale"].severity == "warning"
assert by_agent["agent-stale"].summary == "agent is stale"
assert by_agent["agent-dead"].state == "open"
assert by_agent["agent-dead"].severity == "critical"
assert by_agent["agent-dead"].summary == "agent is dead"
await session.execute(
update(Agent).where(Agent.agent_id == "agent-dead").values(last_seen_at=datetime.now(UTC))
)
await session.commit()
await _tick(sm)
session.expire_all()
resolved = (
await session.execute(
select(Incident).where(
Incident.agent_id == "agent-dead",
Incident.check_id == LIVENESS_CHECK_ID,
)
)
).scalar_one()
assert resolved.state == "resolved"
assert resolved.resolved_at is not None
@pytest.mark.asyncio
async def test_detector_prunes_bounded_history(

57
server/tests/test_events_ingestion.py
View File

@@ -9,7 +9,7 @@ from ._helpers import make_event, make_heartbeat
@pytest.mark.asyncio
async def test_event_accepted_and_state(app_client, auth_headers, session):
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
ev = make_event(status="ok")
ev = make_event(status="ok", output="disk ok")
r = await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": [ev]}, headers=auth_headers
)
@@ -19,6 +19,7 @@ async def test_event_accepted_and_state(app_client, auth_headers, session):
assert n == 1
chk = (await session.execute(select(Check))).scalar_one()
assert chk.status == "ok"
assert chk.summary == ev["output"]
@pytest.mark.asyncio
@@ -64,6 +65,53 @@ async def test_oversize_batch_400(app_client, auth_headers):
assert r.status_code == 400
@pytest.mark.asyncio
async def test_repeated_same_status_not_stored(app_client, auth_headers, session):
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
evs = [make_event(status="ok") for _ in range(5)]
r = await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": evs}, headers=auth_headers
)
assert r.status_code == 202
assert r.json() == {"accepted": 1, "deduplicated": 4}
n = (await session.execute(select(func.count()).select_from(Event))).scalar_one()
assert n == 1
@pytest.mark.asyncio
async def test_status_transitions_stored(app_client, auth_headers, session):
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
seq = [
make_event(status="ok"),
make_event(status="warning", exit_code=1),
make_event(status="warning", exit_code=1),
make_event(status="critical", exit_code=2),
make_event(status="ok"),
]
r = await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": seq}, headers=auth_headers
)
assert r.status_code == 202
assert r.json() == {"accepted": 4, "deduplicated": 1}
n = (await session.execute(select(func.count()).select_from(Event))).scalar_one()
assert n == 4
@pytest.mark.asyncio
async def test_same_status_different_exit_code_stored(app_client, auth_headers, session):
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
seq = [
make_event(status="warning", exit_code=1),
make_event(status="warning", exit_code=3),
]
r = await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": seq}, headers=auth_headers
)
assert r.json() == {"accepted": 2, "deduplicated": 0}
n = (await session.execute(select(func.count()).select_from(Event))).scalar_one()
assert n == 2
@pytest.mark.asyncio
async def test_late_event_does_not_move_state(app_client, auth_headers, session):
from datetime import UTC, datetime, timedelta
@@ -71,7 +119,9 @@ async def test_late_event_does_not_move_state(app_client, auth_headers, session)
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
newer = datetime.now(UTC).replace(microsecond=0)
older = (newer - timedelta(minutes=5)).isoformat()
ev_now = make_event(status="critical", exit_code=2, observed_at=newer.isoformat())
ev_now = make_event(
status="critical", exit_code=2, observed_at=newer.isoformat(), output="current critical"
)
ev_old = make_event(status="ok", exit_code=0, observed_at=older)
await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": [ev_now]}, headers=auth_headers
@@ -81,5 +131,6 @@ async def test_late_event_does_not_move_state(app_client, auth_headers, session)
)
chk = (await session.execute(select(Check))).scalar_one()
assert chk.status == "critical"
assert chk.summary == ev_now["output"]
n = (await session.execute(select(func.count()).select_from(Event))).scalar_one()
assert n == 2
assert n == 1

2
server/tests/test_incident_key.py
View File

@@ -35,7 +35,7 @@ async def test_incident_key_too_long(app_client, auth_headers):
@pytest.mark.asyncio
async def test_events_query_cursor_pagination(app_client, auth_headers):
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
evs = [make_event() for _ in range(5)]
evs = [make_event(check_id=f"chk-{i}") for i in range(5)]
await app_client.post(
"/api/v1/events", json={"agent_id": "agent-1", "events": evs}, headers=auth_headers
)