diff --git a/deploy/docker-compose.showcase.yml b/deploy/docker-compose.showcase.yml index 24a0da0..adc0241 100644 --- a/deploy/docker-compose.showcase.yml +++ b/deploy/docker-compose.showcase.yml @@ -9,7 +9,7 @@ services: MONLET_STALE_AFTER_SEC: "20" MONLET_DEAD_AFTER_SEC: "40" MONLET_DETECTOR_TICK_SEC: "5" - MONLET_EVENTS_RETENTION_MAX_ROWS: "5000" + MONLET_EVENTS_RETENTION_MONTHS: "1" MONLET_OUTBOX_RETENTION_MAX_ROWS: "1000" MONLET_NOTIFIER_DEBUG_ENABLED: "true" MONLET_NOTIFIER_WEBHOOK_ENABLED: "true" @@ -56,6 +56,32 @@ services: agent-mixed: condition: service_started + agent-flap: + image: monlet-showcase-agent + environment: + MONLET_CONFIG: /etc/monlet/agent.toml + volumes: + - ./showcase/agents/flap.toml:/etc/monlet/agent.toml:ro + - ./showcase/checks:/opt/monlet/checks:ro + depends_on: + server: + condition: service_healthy + agent-mixed: + condition: service_started + + agent-timeout: + image: monlet-showcase-agent + environment: + MONLET_CONFIG: /etc/monlet/agent.toml + volumes: + - ./showcase/agents/timeout.toml:/etc/monlet/agent.toml:ro + - ./showcase/checks:/opt/monlet/checks:ro + depends_on: + server: + condition: service_healthy + agent-mixed: + condition: service_started + agent-prometheus-owned: image: monlet-showcase-agent environment: diff --git a/deploy/showcase/agents/flap.toml b/deploy/showcase/agents/flap.toml new file mode 100644 index 0000000..946dc2a --- /dev/null +++ b/deploy/showcase/agents/flap.toml @@ -0,0 +1,25 @@ +hostname = "agent-flap" +state_dir = "/var/lib/monlet-agent" + +[labels] +scenario = "flapping" +role = "showcase" + +[server] +enabled = true +url = "http://server:8000" +token = "monlet-dev-token" +heartbeat_interval = "5s" +batch_interval = "3s" + +[metrics] +enabled = false +listen = "127.0.0.1:9465" + +[[checks]] +id = "flap" +name = "Flapping check (oscillates ok/critical)" +command = "/opt/monlet/checks/flap.sh" +interval = "10s" +timeout = "3s" +dedupe_key = "agent-flap:flap" diff --git a/deploy/showcase/agents/timeout.toml b/deploy/showcase/agents/timeout.toml new file mode 100644 index 0000000..ca14822 --- /dev/null +++ b/deploy/showcase/agents/timeout.toml @@ -0,0 +1,33 @@ +hostname = "agent-timeout" +state_dir = "/var/lib/monlet-agent" + +[labels] +scenario = "timeout" +role = "showcase" + +[server] +enabled = true +url = "http://server:8000" +token = "monlet-dev-token" +heartbeat_interval = "5s" +batch_interval = "3s" + +[metrics] +enabled = false +listen = "127.0.0.1:9465" + +[[checks]] +id = "ok_probe" +name = "Healthy probe" +command = "/opt/monlet/checks/exit_0.sh" +interval = "10s" +timeout = "3s" +dedupe_key = "agent-timeout:ok_probe" + +[[checks]] +id = "slow_probe" +name = "Probe that exceeds the timeout" +command = "/opt/monlet/checks/sleep_long.sh" +interval = "10s" +timeout = "2s" +dedupe_key = "agent-timeout:slow_probe" diff --git a/deploy/showcase/checks/flap.sh b/deploy/showcase/checks/flap.sh new file mode 100755 index 0000000..46a2b53 --- /dev/null +++ b/deploy/showcase/checks/flap.sh @@ -0,0 +1,9 @@ +#!/bin/sh +# Alternates between ok and critical every ~30 seconds to demonstrate flapping. +slot=$(( $(date +%s) / 30 % 2 )) +if [ "$slot" -eq 0 ]; then + echo "critical: flap window down" + exit 2 +fi +echo "ok: flap window up" +exit 0 diff --git a/deploy/showcase/checks/sleep_long.sh b/deploy/showcase/checks/sleep_long.sh new file mode 100755 index 0000000..bb8f957 --- /dev/null +++ b/deploy/showcase/checks/sleep_long.sh @@ -0,0 +1,5 @@ +#!/bin/sh +# Sleeps longer than the configured timeout to demonstrate timeout handling. +sleep 30 +echo "should never reach here" +exit 0 diff --git a/server/.env.example b/server/.env.example index 6e8e10c..e690ba2 100644 --- a/server/.env.example +++ b/server/.env.example @@ -7,7 +7,7 @@ MONLET_STALE_AFTER_SEC=90 MONLET_DEAD_AFTER_SEC=300 MONLET_DETECTOR_TICK_SEC=15 MONLET_ENABLE_DETECTOR=true -MONLET_EVENTS_RETENTION_MAX_ROWS=100000 +MONLET_EVENTS_RETENTION_MONTHS=36 MONLET_OUTBOX_RETENTION_MAX_ROWS=50000 MONLET_ENABLE_NOTIFIER_WORKER=true diff --git a/server/README.md b/server/README.md index a26ba1a..1332049 100644 --- a/server/README.md +++ b/server/README.md @@ -39,7 +39,10 @@ UV_PROJECT_ENVIRONMENT=.venv UV_CACHE_DIR=../.cache/uv uv run fastapi dev monlet | `MONLET_DEAD_AFTER_SEC` | `300` | Dead threshold | | `MONLET_DETECTOR_TICK_SEC` | `15` | Detector tick | | `MONLET_ENABLE_DETECTOR` | `true` | Toggle detector task | -| `MONLET_EVENTS_RETENTION_MAX_ROWS` | `100000` | Maximum retained check-result events; `0` disables pruning | +| `MONLET_EVENTS_RETENTION_MONTHS` | `36` | Months of `events` partitions to keep; `0` disables drop | +| `MONLET_EVENTS_FUTURE_PARTITIONS` | `3` | Future month partitions to pre-create | +| `MONLET_EVENT_DEDUP_RETENTION_DAYS` | `30` | Idempotency window for `event_id` | +| `MONLET_PARTITION_MAINTENANCE_INTERVAL_SEC` | `3600` | Partition create/drop worker interval | | `MONLET_OUTBOX_RETENTION_MAX_ROWS` | `50000` | Maximum retained terminal outbox rows; `0` disables pruning | | `MONLET_ENABLE_NOTIFIER_WORKER` | `true` | Toggle outbox worker | | `MONLET_NOTIFIER_TICK_SEC` | `5` | Worker poll interval | diff --git a/server/alembic/versions/0001_baseline.py b/server/alembic/versions/0001_baseline.py index 2e74195..f745e0e 100644 --- a/server/alembic/versions/0001_baseline.py +++ b/server/alembic/versions/0001_baseline.py @@ -50,10 +50,12 @@ def upgrade() -> None: ) op.execute("CREATE INDEX checks_status_idx ON checks (status);") + # events: partitioned by RANGE(received_at). Partition key is part of PK per Postgres. + # Global idempotency by event_id is enforced via event_ingest_dedup, not this table. op.execute( """ CREATE TABLE events ( - event_id UUID PRIMARY KEY, + event_id UUID NOT NULL, agent_id TEXT NOT NULL, check_id TEXT NOT NULL, observed_at TIMESTAMPTZ NOT NULL, @@ -64,14 +66,30 @@ def upgrade() -> None: output TEXT, output_truncated BOOLEAN NOT NULL DEFAULT FALSE, notifications_enabled BOOLEAN NOT NULL DEFAULT TRUE, - incident_key TEXT NOT NULL - ); + incident_key TEXT NOT NULL, + PRIMARY KEY (event_id, received_at) + ) PARTITION BY RANGE (received_at); """ ) op.execute( "CREATE INDEX events_agent_check_observed_idx ON events (agent_id, check_id, observed_at DESC);" ) op.execute("CREATE INDEX events_observed_at_idx ON events (observed_at DESC);") + op.execute("CREATE INDEX events_received_at_idx ON events (received_at DESC);") + + # Global idempotency table: server enforces event_id uniqueness here, not on the + # partitioned events table. Pruned by retention TTL (event_dedup_retention_days). + op.execute( + """ + CREATE TABLE event_ingest_dedup ( + event_id UUID PRIMARY KEY, + received_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + """ + ) + op.execute( + "CREATE INDEX event_ingest_dedup_received_at_idx ON event_ingest_dedup (received_at);" + ) op.execute( """ @@ -119,6 +137,7 @@ def upgrade() -> None: def downgrade() -> None: op.execute("DROP TABLE IF EXISTS notification_outbox CASCADE;") op.execute("DROP TABLE IF EXISTS incidents CASCADE;") + op.execute("DROP TABLE IF EXISTS event_ingest_dedup CASCADE;") op.execute("DROP TABLE IF EXISTS events CASCADE;") op.execute("DROP TABLE IF EXISTS checks CASCADE;") op.execute("DROP TABLE IF EXISTS agents CASCADE;") diff --git a/server/monlet_server/main.py b/server/monlet_server/main.py index 450380d..e06347c 100644 --- a/server/monlet_server/main.py +++ b/server/monlet_server/main.py @@ -14,19 +14,43 @@ from .middleware.body_limit import BodyLimitMiddleware from .middleware.request_id import RequestIdMiddleware from .services.detector import run_detector from .services.notifier_worker import run_notifier_worker +from .services.partitioning import ensure_event_partitions, run_partition_maintenance from .settings import get_settings log = get_logger("monlet.main") +async def _ensure_partitions_with_retry(engine, attempts: int = 6, base_delay: float = 1.0) -> None: + last_exc: Exception | None = None + for i in range(attempts): + try: + await ensure_event_partitions(engine) + return + except Exception as exc: + last_exc = exc + log.warning("ensure_event_partitions_retry", attempt=i + 1, error=str(exc)) + await asyncio.sleep(min(base_delay * (2**i), 10.0)) + assert last_exc is not None + raise last_exc + + @asynccontextmanager async def lifespan(app: FastAPI): s = get_settings() configure_logging(s.log_level) - get_engine() + engine = get_engine() sm = get_sessionmaker() + # Mandatory: current month partition must exist before accepting writes. + # Retry briefly so a transient DB unavailability at startup (compose race) + # doesn't crash-loop the server. + await _ensure_partitions_with_retry(engine) stop_event = asyncio.Event() tasks: list[asyncio.Task] = [] + # Partition maintenance is independent of detector — must run even when detector is off, + # otherwise events_future_partitions runs out and ingest fails. + tasks.append( + asyncio.create_task(run_partition_maintenance(sm, stop_event), name="partition_maintenance") + ) if s.enable_detector: tasks.append(asyncio.create_task(run_detector(sm, stop_event), name="detector")) if s.enable_notifier_worker: diff --git a/server/monlet_server/models.py b/server/monlet_server/models.py index a810810..acb5b3d 100644 --- a/server/monlet_server/models.py +++ b/server/monlet_server/models.py @@ -85,10 +85,11 @@ class Check(Base): class Event(Base): __tablename__ = "events" - event_id: Mapped[UUID] = mapped_column(_uuid(), primary_key=True) + event_id: Mapped[UUID] = mapped_column(_uuid(), nullable=False) agent_id: Mapped[str] = mapped_column(Text, nullable=False) check_id: Mapped[str] = mapped_column(Text, nullable=False) observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + # Partition key (RANGE) on Postgres; composite PK = (event_id, received_at). received_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) @@ -103,6 +104,7 @@ class Event(Base): incident_key: Mapped[str] = mapped_column(Text, nullable=False) __table_args__ = ( + PrimaryKeyConstraint("event_id", "received_at", name="events_pkey"), CheckConstraint( "status IN ('ok','warning','critical','unknown')", name="events_status_chk" ), @@ -114,9 +116,21 @@ class Event(Base): text("observed_at DESC"), ), Index("events_observed_at_idx", text("observed_at DESC")), + Index("events_received_at_idx", text("received_at DESC")), ) +class EventIngestDedup(Base): + __tablename__ = "event_ingest_dedup" + + event_id: Mapped[UUID] = mapped_column(_uuid(), primary_key=True) + received_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = (Index("event_ingest_dedup_received_at_idx", "received_at"),) + + class Incident(Base): __tablename__ = "incidents" diff --git a/server/monlet_server/services/detector.py b/server/monlet_server/services/detector.py index 75832b2..6c3c410 100644 --- a/server/monlet_server/services/detector.py +++ b/server/monlet_server/services/detector.py @@ -1,6 +1,6 @@ import asyncio from datetime import UTC, datetime, timedelta -from uuid import uuid4 +from uuid import UUID, uuid4, uuid7 from sqlalchemy import delete, func, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert @@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker from .. import metrics from ..logging_config import get_logger -from ..models import Agent, Event, Incident, NotificationOutbox +from ..models import Agent, Check, Event, EventIngestDedup, Incident, NotificationOutbox from ..settings import get_settings log = get_logger("monlet.detector") @@ -25,6 +25,10 @@ def _liveness_severity(status: str) -> str: return "critical" if status == "dead" else "warning" +_LIVENESS_CHECK_STATUS = {"alive": "ok", "stale": "warning", "dead": "critical"} +_LIVENESS_EXIT_CODE = {"ok": 0, "warning": 1, "critical": 2} + + async def _enqueue_liveness_outbox( session, incident: Incident, @@ -68,7 +72,9 @@ async def _enqueue_liveness_outbox( ) -async def _open_liveness_incident(session, agent: Agent, status: str, now: datetime) -> None: +async def _open_liveness_incident( + session, agent: Agent, status: str, now: datetime, event_id: UUID +) -> None: severity = _liveness_severity(status) summary = f"agent is {status}" incident_key = _liveness_key(agent.agent_id) @@ -84,7 +90,7 @@ async def _open_liveness_incident(session, agent: Agent, status: str, now: datet severity=severity, opened_at=now, summary=summary, - last_event_id=uuid4(), + last_event_id=event_id, ) .on_conflict_do_nothing( index_elements=[Incident.incident_key], @@ -115,13 +121,13 @@ async def _open_liveness_incident(session, agent: Agent, status: str, now: datet if status == "dead" and inc.severity != "critical": inc.severity = "critical" inc.summary = summary - inc.last_event_id = uuid4() + inc.last_event_id = event_id 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: +async def _resolve_liveness_incident(session, agent: Agent, now: datetime, event_id: UUID) -> None: inc = ( await session.execute( select(Incident).where( @@ -134,7 +140,7 @@ async def _resolve_liveness_incident(session, agent: Agent, now: datetime) -> No return inc.state = "resolved" inc.resolved_at = now - inc.last_event_id = uuid4() + inc.last_event_id = event_id metrics.incidents_resolved_total.inc() await _enqueue_liveness_outbox( session, @@ -148,15 +154,83 @@ async def _resolve_liveness_incident(session, agent: Agent, now: datetime) -> No ) +async def _apply_liveness(session, agent: Agent, next_status: str, now: datetime) -> None: + """Server-owned liveness check: maintain Check + Event-on-transition + Incident lifecycle.""" + liveness_status = _LIVENESS_CHECK_STATUS[next_status] + exit_code = _LIVENESS_EXIT_CODE[liveness_status] + summary = f"agent is {next_status}" + incident_key = _liveness_key(agent.agent_id) + + cur = ( + await session.execute( + select(Check).where( + Check.agent_id == agent.agent_id, Check.check_id == LIVENESS_CHECK_ID + ) + ) + ).scalar_one_or_none() + changed = cur is None or cur.status != liveness_status + + # Liveness events flow through /events/query which validates event_id as UUIDv7. + event_id = cur.last_event_id if cur and not changed else uuid7() + if changed: + session.add( + Event( + event_id=event_id, + agent_id=agent.agent_id, + check_id=LIVENESS_CHECK_ID, + observed_at=now, + status=liveness_status, + exit_code=exit_code, + duration_ms=0, + output=summary, + output_truncated=False, + notifications_enabled=True, + incident_key=incident_key, + ) + ) + + if cur is None: + session.add( + Check( + agent_id=agent.agent_id, + check_id=LIVENESS_CHECK_ID, + status=liveness_status, + exit_code=exit_code, + last_observed_at=now, + last_event_id=event_id, + incident_key=incident_key, + summary=summary, + ) + ) + else: + cur.last_observed_at = now + cur.summary = summary + if changed: + cur.status = liveness_status + cur.exit_code = exit_code + cur.last_event_id = event_id + cur.incident_key = incident_key + + if next_status in ("stale", "dead"): + await _open_liveness_incident(session, agent, next_status, now, event_id) + else: + await _resolve_liveness_incident(session, agent, now, event_id) + + async def _prune_history(session) -> None: s = get_settings() - if s.events_retention_max_rows > 0: - keep_events = ( - select(Event.event_id) - .order_by(Event.received_at.desc(), Event.event_id.desc()) - .limit(s.events_retention_max_rows) + if s.event_dedup_retention_days > 0: + cutoff = datetime.now(UTC) - timedelta(days=s.event_dedup_retention_days) + batch = max(s.event_dedup_prune_batch_size, 1) + # Bounded DELETE: keep each tick's prune transaction short to avoid bloat and + # long-running locks on a large dedup table. + sub = ( + select(EventIngestDedup.event_id) + .where(EventIngestDedup.received_at < cutoff) + .limit(batch) + .scalar_subquery() ) - await session.execute(delete(Event).where(Event.event_id.not_in(keep_events))) + await session.execute(delete(EventIngestDedup).where(EventIngestDedup.event_id.in_(sub))) if s.outbox_retention_max_rows > 0: keep_outbox = ( @@ -188,10 +262,7 @@ async def _tick(sm: async_sessionmaker) -> None: 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 _apply_liveness(session, agent, next_status, now) await _prune_history(session) await session.commit() diff --git a/server/monlet_server/services/ingestion.py b/server/monlet_server/services/ingestion.py index 2830584..a62e456 100644 --- a/server/monlet_server/services/ingestion.py +++ b/server/monlet_server/services/ingestion.py @@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from .. import metrics from ..logging_config import get_logger -from ..models import Agent, Check, Event, Incident, NotificationOutbox +from ..models import Agent, Check, Event, EventIngestDedup, Incident, NotificationOutbox from ..redaction import redact from ..schemas import CheckResultEvent, HeartbeatRequest from ..settings import get_settings @@ -155,26 +155,45 @@ async def ingest_event( agent_id: str, ev: CheckResultEvent, ) -> bool: - """Return True if a new change row was written, False if not (duplicate/unchanged/late).""" + """Return True if a new change row was written, False if not (duplicate/unchanged/late). + + Idempotency is enforced first via the standalone event_ingest_dedup table — a + duplicate event_id never touches checks/incidents/outbox/events. The full pipeline + (dedup + events + check + incident + outbox) commits as a single transaction by + the caller, so a partial failure cannot leave a dedup row without its side-effects. + """ 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) + event_uuid = UUID(ev.event_id) + # Order matters: late check before dedup insert so that repeated late events stay + # observable as `late`, not silently collapsed into `deduplicated`. 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 - changed = cur is None or cur.status != ev.status or cur.exit_code != ev.exit_code - if is_late: metrics.events_total.labels(result="late").inc() return False + dedup_stmt = ( + pg_insert(EventIngestDedup) + .values(event_id=event_uuid, received_at=datetime.now(UTC)) + .on_conflict_do_nothing(index_elements=[EventIngestDedup.event_id]) + .returning(EventIngestDedup.event_id) + ) + if (await session.execute(dedup_stmt)).scalar_one_or_none() is None: + metrics.events_total.labels(result="deduplicated").inc() + return False + + changed = cur is None or cur.status != ev.status or cur.exit_code != ev.exit_code + if changed: stmt = pg_insert(Event).values( - event_id=UUID(ev.event_id), + event_id=event_uuid, agent_id=agent_id, check_id=ev.check_id, observed_at=observed, @@ -186,14 +205,7 @@ async def ingest_event( 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 + await session.execute(stmt) metrics.events_total.labels(result="accepted").inc() else: metrics.events_total.labels(result="unchanged").inc() @@ -218,7 +230,7 @@ async def ingest_event( "summary": check_summary, } if changed: - update_set["last_event_id"] = UUID(ev.event_id) + update_set["last_event_id"] = event_uuid update_set["incident_key"] = incident_key chk_stmt = pg_insert(Check).values(**check_values) chk_stmt = chk_stmt.on_conflict_do_update( diff --git a/server/monlet_server/services/partitioning.py b/server/monlet_server/services/partitioning.py new file mode 100644 index 0000000..0082c11 --- /dev/null +++ b/server/monlet_server/services/partitioning.py @@ -0,0 +1,143 @@ +"""Monthly partition maintenance for `events` (PARTITION BY RANGE(received_at)). + +Idempotent: creates current + N future month partitions and drops old ones beyond +the retention window. Names are derived from year/month, so they cannot inject SQL. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, date, datetime + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker + +from ..logging_config import get_logger +from ..settings import get_settings + +log = get_logger("monlet.partitioning") + + +def _month_start(d: date) -> date: + return date(d.year, d.month, 1) + + +def _next_month(d: date) -> date: + year = d.year + (1 if d.month == 12 else 0) + month = 1 if d.month == 12 else d.month + 1 + return date(year, month, 1) + + +def _shift_months(d: date, n: int) -> date: + y, m = d.year, d.month + total = y * 12 + (m - 1) + n + return date(total // 12, (total % 12) + 1, 1) + + +def _partition_name(d: date) -> str: + return f"events_p{d.year:04d}_{d.month:02d}" + + +async def ensure_event_partitions(engine_or_sm: AsyncEngine | async_sessionmaker) -> None: + """Create current + N future month partitions for events. Idempotent.""" + s = get_settings() + today = _month_start(datetime.now(UTC).date()) + months = [_shift_months(today, i) for i in range(0, max(s.events_future_partitions, 0) + 1)] + + statements: list[str] = [] + for start in months: + end = _next_month(start) + # Explicit UTC qualifier — Postgres otherwise interprets bare date literals via + # the session TimeZone, which would shift partition bounds on non-UTC servers. + statements.append( + f"CREATE TABLE IF NOT EXISTS {_partition_name(start)} " + f"PARTITION OF events FOR VALUES FROM ('{start.isoformat()} UTC') " + f"TO ('{end.isoformat()} UTC');" + ) + + async with _connect(engine_or_sm) as conn: + for stmt in statements: + await conn.execute(text(stmt)) + + +async def drop_old_event_partitions(engine_or_sm: AsyncEngine | async_sessionmaker) -> None: + """Drop month partitions older than events_retention_months. + + Each DROP runs in its own transaction so the ACCESS EXCLUSIVE lock on parent + `events` is held only for one partition at a time and ingest is not blocked + by a long maintenance batch. + """ + s = get_settings() + if s.events_retention_months <= 0: + return + cutoff = _shift_months(_month_start(datetime.now(UTC).date()), -s.events_retention_months) + + engine = _resolve_engine(engine_or_sm) + async with engine.connect() as conn: + rows = ( + await conn.execute( + text( + """ + SELECT c.relname + FROM pg_inherits i + JOIN pg_class c ON c.oid = i.inhrelid + JOIN pg_class p ON p.oid = i.inhparent + WHERE p.relname = 'events' + """ + ) + ) + ).all() + + to_drop: list[str] = [] + for (name,) in rows: + if not name.startswith("events_p"): + continue + try: + y = int(name[8:12]) + m = int(name[13:15]) + except ValueError: + continue + if date(y, m, 1) < cutoff: + to_drop.append(name) + + for name in to_drop: + async with engine.begin() as conn: + await conn.execute(text(f"DROP TABLE IF EXISTS {name};")) + log.info("event_partition_dropped", name=name) + + +def _resolve_engine(engine_or_sm: AsyncEngine | async_sessionmaker) -> AsyncEngine: + if isinstance(engine_or_sm, async_sessionmaker): + return engine_or_sm.kw["bind"] + return engine_or_sm + + +def _connect(engine_or_sm): + """Return an async ctx for a transactional connection from either engine or sessionmaker.""" + return _resolve_engine(engine_or_sm).begin() + + +async def run_partition_maintenance(sm: async_sessionmaker, stop_event: asyncio.Event) -> None: + """Independent worker — runs even when MONLET_ENABLE_DETECTOR=false.""" + interval = max(get_settings().partition_maintenance_interval_sec, 60) + while not stop_event.is_set(): + try: + await ensure_event_partitions(sm) + await drop_old_event_partitions(sm) + except Exception as exc: + log.warning("partition_maintenance_failed", error=str(exc)) + try: + await asyncio.wait_for(stop_event.wait(), timeout=interval) + except TimeoutError: + continue + + +# Re-export helpers for tests. +__all__ = [ + "run_partition_maintenance", + "ensure_event_partitions", + "drop_old_event_partitions", + "_partition_name", + "_shift_months", + "_month_start", +] diff --git a/server/monlet_server/settings.py b/server/monlet_server/settings.py index 59dc325..74760a2 100644 --- a/server/monlet_server/settings.py +++ b/server/monlet_server/settings.py @@ -20,7 +20,11 @@ class Settings(BaseSettings): port: int = 8000 enable_detector: bool = True enable_notifier_worker: bool = True - events_retention_max_rows: int = 20_000 + events_retention_months: int = 36 + events_future_partitions: int = 3 + event_dedup_retention_days: int = 30 + event_dedup_prune_batch_size: int = 5000 + partition_maintenance_interval_sec: int = 3600 outbox_retention_max_rows: int = 50_000 notifier_tick_sec: int = 5 notifier_batch_size: int = 20 diff --git a/server/tests/conftest.py b/server/tests/conftest.py index 42174ef..6465c10 100644 --- a/server/tests/conftest.py +++ b/server/tests/conftest.py @@ -58,6 +58,20 @@ def _apply_migrations(database_url: str) -> Iterator[None]: cfg.set_main_option("script_location", os.path.join(SERVER_ROOT, "alembic")) cfg.set_main_option("sqlalchemy.url", get_settings().sync_database_url) command.upgrade(cfg, "head") + + # Partitions for `events` are created at runtime, not by the migration. + async def _bootstrap() -> None: + from sqlalchemy.ext.asyncio import create_async_engine + + from monlet_server.services.partitioning import ensure_event_partitions + + eng = create_async_engine(database_url, future=True) + try: + await ensure_event_partitions(eng) + finally: + await eng.dispose() + + asyncio.run(_bootstrap()) yield @@ -83,7 +97,7 @@ async def _truncate_tables(engine) -> AsyncIterator[None]: async with engine.begin() as conn: await conn.execute( text( - "TRUNCATE TABLE notification_outbox, incidents, events, checks, agents RESTART IDENTITY CASCADE" + "TRUNCATE TABLE notification_outbox, incidents, events, event_ingest_dedup, checks, agents RESTART IDENTITY CASCADE" ) ) diff --git a/server/tests/test_constraints.py b/server/tests/test_constraints.py index c463604..1bde6e0 100644 --- a/server/tests/test_constraints.py +++ b/server/tests/test_constraints.py @@ -9,26 +9,18 @@ from monlet_server.models import Incident @pytest.mark.asyncio -async def test_event_id_conflict_do_nothing(engine): +async def test_event_id_dedup_conflict_do_nothing(engine): + """Global idempotency is enforced via event_ingest_dedup, not the partitioned events table.""" eid = str(uuid4()) async with engine.begin() as conn: await conn.execute( - text( - "INSERT INTO agents (agent_id, hostname, features, status, last_seen_at) " - "VALUES ('a','h',jsonb_build_object('push', true, 'metrics', false),'alive', now())" - ) - ) - await conn.execute( - text( - "INSERT INTO events (event_id, agent_id, check_id, observed_at, status, exit_code, duration_ms, incident_key)" - " VALUES (:e,'a','c', now(),'ok',0,1,'a:c')" - ), + text("INSERT INTO event_ingest_dedup (event_id) VALUES (:e)"), {"e": eid}, ) r = await conn.execute( text( - "INSERT INTO events (event_id, agent_id, check_id, observed_at, status, exit_code, duration_ms, incident_key)" - " VALUES (:e,'a','c', now(),'ok',0,1,'a:c') ON CONFLICT (event_id) DO NOTHING RETURNING event_id" + "INSERT INTO event_ingest_dedup (event_id) VALUES (:e) " + "ON CONFLICT (event_id) DO NOTHING RETURNING event_id" ), {"e": eid}, ) diff --git a/server/tests/test_detector.py b/server/tests/test_detector.py index 7cb85c9..99fc769 100644 --- a/server/tests/test_detector.py +++ b/server/tests/test_detector.py @@ -5,7 +5,7 @@ import pytest 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.models import Agent, Check, Event, Incident, NotificationOutbox from monlet_server.services.detector import LIVENESS_CHECK_ID, _tick from monlet_server.settings import reset_settings_cache @@ -82,22 +82,75 @@ async def test_detector_transitions(app_client, auth_headers, engine, session): @pytest.mark.asyncio -async def test_detector_prunes_bounded_history( - app_client, auth_headers, engine, session, monkeypatch -): - monkeypatch.setenv("MONLET_EVENTS_RETENTION_MAX_ROWS", "2") +async def test_detector_owns_liveness_check_and_events(app_client, auth_headers, engine, session): + await app_client.post( + "/api/v1/heartbeat", json=make_heartbeat("agent-flap"), headers=auth_headers + ) + sm = async_sessionmaker(engine, expire_on_commit=False) + + # Tick 1: alive → Check exists with status=ok, one event. + await _tick(sm) + session.expire_all() + check = ( + await session.execute( + select(Check).where(Check.agent_id == "agent-flap", Check.check_id == LIVENESS_CHECK_ID) + ) + ).scalar_one() + assert check.status == "ok" + n_events = ( + await session.execute( + select(func.count()) + .select_from(Event) + .where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID) + ) + ).scalar_one() + assert n_events == 1 + + # Tick 2: still alive → no new event. + await _tick(sm) + session.expire_all() + n_events = ( + await session.execute( + select(func.count()) + .select_from(Event) + .where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID) + ) + ).scalar_one() + assert n_events == 1 + + # Push agent into dead → transition event written, Check updates. + await session.execute( + update(Agent) + .where(Agent.agent_id == "agent-flap") + .values(last_seen_at=datetime.now(UTC) - timedelta(minutes=10)) + ) + await session.commit() + await _tick(sm) + session.expire_all() + check = ( + await session.execute( + select(Check).where(Check.agent_id == "agent-flap", Check.check_id == LIVENESS_CHECK_ID) + ) + ).scalar_one() + assert check.status == "critical" + n_events = ( + await session.execute( + select(func.count()) + .select_from(Event) + .where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID) + ) + ).scalar_one() + assert n_events == 2 + + +@pytest.mark.asyncio +async def test_detector_prunes_outbox(app_client, auth_headers, engine, session, monkeypatch): monkeypatch.setenv("MONLET_OUTBOX_RETENTION_MAX_ROWS", "2") reset_settings_cache() try: await app_client.post( "/api/v1/heartbeat", json=make_heartbeat("agent-prune"), headers=auth_headers ) - events = [make_event(check_id="prune_check") for _ in range(5)] - await app_client.post( - "/api/v1/events", - json={"agent_id": "agent-prune", "events": events}, - headers=auth_headers, - ) critical = make_event(check_id="critical_prune", status="critical", exit_code=2) await app_client.post( "/api/v1/events", @@ -129,7 +182,6 @@ async def test_detector_prunes_bounded_history( sm = async_sessionmaker(engine, expire_on_commit=False) await _tick(sm) - assert (await session.execute(select(func.count()).select_from(Event))).scalar_one() == 2 terminal_outbox = ( await session.execute( select(func.count()) diff --git a/server/tests/test_partitioning.py b/server/tests/test_partitioning.py new file mode 100644 index 0000000..c2dde02 --- /dev/null +++ b/server/tests/test_partitioning.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +from datetime import UTC, date, datetime, timedelta + +import pytest +from sqlalchemy import text + +from monlet_server.services.partitioning import ( + _month_start, + _partition_name, + _shift_months, + drop_old_event_partitions, + ensure_event_partitions, +) +from monlet_server.settings import reset_settings_cache + + +async def _partitions(engine) -> set[str]: + async with engine.begin() as conn: + rows = ( + await conn.execute( + text( + "SELECT c.relname FROM pg_inherits i " + "JOIN pg_class c ON c.oid=i.inhrelid " + "JOIN pg_class p ON p.oid=i.inhparent " + "WHERE p.relname='events'" + ) + ) + ).all() + return {r[0] for r in rows} + + +@pytest.mark.asyncio +async def test_ensure_creates_current_and_future_partitions(engine, monkeypatch): + monkeypatch.setenv("MONLET_EVENTS_FUTURE_PARTITIONS", "2") + reset_settings_cache() + try: + await ensure_event_partitions(engine) + today = _month_start(datetime.now(UTC).date()) + for i in range(0, 3): + assert _partition_name(_shift_months(today, i)) in await _partitions(engine) + finally: + reset_settings_cache() + + +@pytest.mark.asyncio +async def test_drop_old_partitions_uses_retention(engine, monkeypatch): + monkeypatch.setenv("MONLET_EVENTS_RETENTION_MONTHS", "1") + reset_settings_cache() + try: + # Create an old partition directly. + old = date(2020, 1, 1) + name = _partition_name(old) + end = _shift_months(old, 1) + async with engine.begin() as conn: + await conn.execute( + text( + f"CREATE TABLE IF NOT EXISTS {name} PARTITION OF events " + f"FOR VALUES FROM ('{old.isoformat()} UTC') TO ('{end.isoformat()} UTC');" + ) + ) + assert name in await _partitions(engine) + await drop_old_event_partitions(engine) + assert name not in await _partitions(engine) + finally: + reset_settings_cache() + + +@pytest.mark.asyncio +async def test_dedup_prevents_replay_via_ingest(app_client, auth_headers, session): + from sqlalchemy import func, select + + from monlet_server.models import Event + + from ._helpers import make_event, make_heartbeat + + await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers) + ev = make_event(status="critical", exit_code=2) + + r1 = await app_client.post( + "/api/v1/events", json={"agent_id": "agent-1", "events": [ev]}, headers=auth_headers + ) + assert r1.json() == {"accepted": 1, "deduplicated": 0} + + # Replay the exact same event_id long after — dedup must block ingestion entirely, + # never touching checks/incidents/outbox/events. + r2 = await app_client.post( + "/api/v1/events", json={"agent_id": "agent-1", "events": [ev]}, headers=auth_headers + ) + assert r2.json() == {"accepted": 0, "deduplicated": 1} + + n = (await session.execute(select(func.count()).select_from(Event))).scalar_one() + assert n == 1 + + +@pytest.mark.asyncio +async def test_late_dedup_does_not_keep_replay_doors_open(engine, monkeypatch): + """Dedup retention TTL must be at least documented agent spool window.""" + from uuid import uuid4 + + from sqlalchemy import func, select + from sqlalchemy.ext.asyncio import async_sessionmaker + + from monlet_server.models import EventIngestDedup + from monlet_server.services.detector import _prune_history + + monkeypatch.setenv("MONLET_EVENT_DEDUP_RETENTION_DAYS", "1") + reset_settings_cache() + try: + sm = async_sessionmaker(engine, expire_on_commit=False) + async with sm() as s: + old = datetime.now(UTC) - timedelta(days=5) + s.add(EventIngestDedup(event_id=uuid4(), received_at=old)) + s.add(EventIngestDedup(event_id=uuid4(), received_at=datetime.now(UTC))) + await s.commit() + + async with sm() as s: + await _prune_history(s) + await s.commit() + + async with sm() as s: + n = (await s.execute(select(func.count()).select_from(EventIngestDedup))).scalar_one() + assert n == 1 + finally: + reset_settings_cache() diff --git a/web/src/app/agents/[agent_id]/page.tsx b/web/src/app/agents/[agent_id]/page.tsx index 48e0fc7..18b0b5d 100644 --- a/web/src/app/agents/[agent_id]/page.tsx +++ b/web/src/app/agents/[agent_id]/page.tsx @@ -1,13 +1,14 @@ import { AgentDetailBrowser } from "@/components/AgentDetailBrowser"; import { ErrorPanel } from "@/components/DataPanel"; import { ApiError, api } from "@/lib/api"; -import { withLivenessEvents } from "@/lib/check-groups"; import { loadChecks } from "@/lib/loaders"; +import { normalizePageSize } from "@/lib/pagination"; export const dynamic = "force-dynamic"; type SortKey = "severity" | "check_id" | "last_seen"; type TabKey = "checks" | "events"; +type EventsPage = Awaited>; function normalizeSort(sort?: string): SortKey { return sort === "check_id" || sort === "last_seen" ? sort : "severity"; @@ -21,22 +22,35 @@ export default async function AgentDetailPage({ params, searchParams, }: { - params: Promise<{ agent_id: string }>; - searchParams: Promise<{ sort?: string; tab?: string }>; + params: Promise<{ + agent_id: string; + }>; + searchParams: Promise<{ + sort?: string; + tab?: string; + events_limit?: string; + cursor?: string; + }>; }) { const { agent_id } = await params; const sp = await searchParams; const sort = normalizeSort(sp.sort); const tab = normalizeTab(sp.tab); + const eventsLimit = normalizePageSize(sp.events_limit, 50); + const eventsCursor = sp.cursor; - let data; + let agent; + let checks; + let events: EventsPage = { items: [], next_cursor: null }; try { - const [agent, checks, events] = await Promise.all([ - api.getAgent(agent_id), - loadChecks(agent_id), - api.queryEvents({ agent_id, limit: 50 }).then((page) => page.items), - ]); - data = { agent, checks, events: withLivenessEvents([agent], events, { agentId: agent_id }) }; + [agent, checks] = await Promise.all([api.getAgent(agent_id), loadChecks(agent_id)]); + if (tab === "events") { + events = await api.queryEvents({ + agent_id, + limit: eventsLimit, + cursor: eventsCursor, + }); + } } catch (e) { if (e instanceof ApiError && e.status === 404) { return
Agent not found.
; @@ -45,10 +59,13 @@ export default async function AgentDetailPage({ } return ( ); } diff --git a/web/src/app/api/poll/agent/route.ts b/web/src/app/api/poll/agent/route.ts index ccfa303..595c1be 100644 --- a/web/src/app/api/poll/agent/route.ts +++ b/web/src/app/api/poll/agent/route.ts @@ -1,7 +1,6 @@ import { NextResponse } from "next/server"; import { ApiError, api } from "@/lib/api"; -import { withLivenessEvents } from "@/lib/check-groups"; import { loadChecks } from "@/lib/loaders"; import { jsonError } from "@/lib/poll-response"; @@ -19,16 +18,8 @@ export async function GET(request: Request) { } try { - const [agent, checks, events] = await Promise.all([ - api.getAgent(agentId), - loadChecks(agentId), - api.queryEvents({ agent_id: agentId, limit: 50 }).then((page) => page.items), - ]); - return NextResponse.json({ - agent, - checks, - events: withLivenessEvents([agent], events, { agentId }), - }); + const [agent, checks] = await Promise.all([api.getAgent(agentId), loadChecks(agentId)]); + return NextResponse.json({ agent, checks }); } catch (e) { if (e instanceof ApiError && e.status === 404) { return NextResponse.json( diff --git a/web/src/app/api/poll/events/route.ts b/web/src/app/api/poll/events/route.ts index 281f7e7..6bef416 100644 --- a/web/src/app/api/poll/events/route.ts +++ b/web/src/app/api/poll/events/route.ts @@ -1,20 +1,21 @@ import { NextResponse } from "next/server"; -import { loadEventsWithLiveness } from "@/lib/loaders"; +import { loadEvents } from "@/lib/loaders"; +import { normalizePageSize } from "@/lib/pagination"; import { jsonError } from "@/lib/poll-response"; export const dynamic = "force-dynamic"; export async function GET(request: Request) { const url = new URL(request.url); - const limit = Number(url.searchParams.get("limit") ?? "100"); + const limit = normalizePageSize(url.searchParams.get("limit"), 50); try { - const data = await loadEventsWithLiveness({ + const data = await loadEvents({ agentId: url.searchParams.get("agent_id") ?? undefined, checkId: url.searchParams.get("check_id") ?? undefined, cursor: url.searchParams.get("cursor") ?? undefined, - limit: Number.isFinite(limit) ? limit : 100, + limit, }); return NextResponse.json(data); } catch (e) { diff --git a/web/src/app/api/poll/incidents/route.ts b/web/src/app/api/poll/incidents/route.ts index ff96949..86cbf9e 100644 --- a/web/src/app/api/poll/incidents/route.ts +++ b/web/src/app/api/poll/incidents/route.ts @@ -11,8 +11,8 @@ export async function GET(request: Request) { const state = stateParam === "open" || stateParam === "resolved" ? stateParam : undefined; try { - const data = await loadIncidents(state); - return NextResponse.json(data); + const items = await loadIncidents(state); + return NextResponse.json({ items }); } catch (e) { return jsonError(e); } diff --git a/web/src/app/api/poll/outbox/route.ts b/web/src/app/api/poll/outbox/route.ts index 65d9222..d61b90b 100644 --- a/web/src/app/api/poll/outbox/route.ts +++ b/web/src/app/api/poll/outbox/route.ts @@ -1,18 +1,14 @@ import { NextResponse } from "next/server"; -import { api } from "@/lib/api"; +import { loadOutboxAll } from "@/lib/loaders"; import { jsonError } from "@/lib/poll-response"; export const dynamic = "force-dynamic"; -export async function GET(request: Request) { - const url = new URL(request.url); - +export async function GET() { try { - const data = await api.listOutbox({ - cursor: url.searchParams.get("cursor") ?? undefined, - }); - return NextResponse.json(data); + const items = await loadOutboxAll(); + return NextResponse.json({ items }); } catch (e) { return jsonError(e); } diff --git a/web/src/app/checks/[check_id]/page.tsx b/web/src/app/checks/[check_id]/page.tsx index 789eb9b..350fa93 100644 --- a/web/src/app/checks/[check_id]/page.tsx +++ b/web/src/app/checks/[check_id]/page.tsx @@ -1,8 +1,8 @@ import { CheckDetailBrowser } from "@/components/CheckDetailBrowser"; import { ErrorPanel } from "@/components/DataPanel"; import { api } from "@/lib/api"; -import { withLivenessEvents } from "@/lib/check-groups"; import { loadInventory } from "@/lib/loaders"; +import { normalizePageSize } from "@/lib/pagination"; export const dynamic = "force-dynamic"; @@ -18,22 +18,24 @@ export default async function CheckDetailPage({ searchParams, }: { params: Promise<{ check_id: string }>; - searchParams: Promise<{ tab?: string }>; + searchParams: Promise<{ tab?: string; events_limit?: string; cursor?: string }>; }) { const { check_id } = await params; const sp = await searchParams; const tab = normalizeTab(sp.tab); + const eventsLimit = normalizePageSize(sp.events_limit, 50); + const eventsCursor = sp.cursor; let data; let events: EventsPage = { items: [], next_cursor: null }; try { data = await loadInventory(); if (tab === "events") { - const rawEvents = await api.queryEvents({ check_id, limit: 50 }); - events = { - ...rawEvents, - items: withLivenessEvents(data.agents, rawEvents.items, { checkId: check_id }), - }; + events = await api.queryEvents({ + check_id, + limit: eventsLimit, + cursor: eventsCursor, + }); } } catch (e) { return ; @@ -41,11 +43,13 @@ export default async function CheckDetailPage({ return ( ); } diff --git a/web/src/app/events/page.tsx b/web/src/app/events/page.tsx index caf628a..6e5b42b 100644 --- a/web/src/app/events/page.tsx +++ b/web/src/app/events/page.tsx @@ -1,32 +1,41 @@ import { ErrorPanel } from "@/components/DataPanel"; import { EventsBrowser } from "@/components/EventsBrowser"; -import { loadEventsWithLiveness } from "@/lib/loaders"; +import { loadEvents } from "@/lib/loaders"; +import { normalizePageSize } from "@/lib/pagination"; export const dynamic = "force-dynamic"; export default async function EventsPage({ searchParams, }: { - searchParams: Promise<{ cursor?: string; agent_id?: string; check_id?: string }>; + searchParams: Promise<{ + cursor?: string; + agent_id?: string; + check_id?: string; + limit?: string; + }>; }) { const sp = await searchParams; + const limit = normalizePageSize(sp.limit); let data; try { - data = await loadEventsWithLiveness({ + data = await loadEvents({ agentId: sp.agent_id, checkId: sp.check_id, cursor: sp.cursor, + limit, }); } catch (e) { return ; } return ( ); } diff --git a/web/src/app/incidents/page.tsx b/web/src/app/incidents/page.tsx index 74ad4fa..eef1aa6 100644 --- a/web/src/app/incidents/page.tsx +++ b/web/src/app/incidents/page.tsx @@ -11,17 +11,17 @@ export default async function IncidentsPage({ }) { const sp = await searchParams; const state = sp.state === "open" || sp.state === "resolved" ? sp.state : undefined; - let data; + let items; let inventory; try { - [data, inventory] = await Promise.all([loadIncidents(state), loadInventory()]); + [items, inventory] = await Promise.all([loadIncidents(state), loadInventory()]); } catch (e) { return ; } return ( diff --git a/web/src/app/outbox/page.tsx b/web/src/app/outbox/page.tsx index 47d9047..c3bf351 100644 --- a/web/src/app/outbox/page.tsx +++ b/web/src/app/outbox/page.tsx @@ -1,20 +1,15 @@ import { ErrorPanel } from "@/components/DataPanel"; import { OutboxBrowser } from "@/components/OutboxBrowser"; -import { api } from "@/lib/api"; +import { loadOutboxAll } from "@/lib/loaders"; export const dynamic = "force-dynamic"; -export default async function OutboxPage({ - searchParams, -}: { - searchParams: Promise<{ cursor?: string }>; -}) { - const sp = await searchParams; - let data; +export default async function OutboxPage() { + let items; try { - data = await api.listOutbox({ cursor: sp.cursor }); + items = await loadOutboxAll(); } catch (e) { return ; } - return ; + return ; } diff --git a/web/src/components/AgentDetailBrowser.tsx b/web/src/components/AgentDetailBrowser.tsx index e1773db..b9a3694 100644 --- a/web/src/components/AgentDetailBrowser.tsx +++ b/web/src/components/AgentDetailBrowser.tsx @@ -1,28 +1,37 @@ "use client"; import Link from "next/link"; -import { useMemo } from "react"; +import { useRouter } from "next/navigation"; +import { useMemo, useState } from "react"; import { AgentFeatures } from "@/components/AgentFeatures"; import { CheckEventsLink } from "@/components/CheckEventsLink"; -import { Td, Th } from "@/components/DataPanel"; +import { ClientPager } from "@/components/ClientPager"; +import { EmptyPanel, SummaryCell, Td, Th } from "@/components/DataPanel"; +import { DetailTabs } from "@/components/DetailTabs"; import { LabelChips } from "@/components/Labels"; +import { PageSizeSelect } from "@/components/PageSizeSelect"; +import { Pager } from "@/components/Pager"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; -import { withLivenessChecks } from "@/lib/check-groups"; import { statusBadgeClass } from "@/lib/format"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; type Agent = components["schemas"]["Agent"]; type CheckState = components["schemas"]["CheckState"]; type StoredEvent = components["schemas"]["StoredCheckResultEvent"]; +type EventsPageData = { items: StoredEvent[]; next_cursor?: string | null }; type SortKey = "severity" | "check_id" | "last_seen"; type TabKey = "checks" | "events"; -type AgentDetailData = { - agent: Agent; - checks: CheckState[]; - events: StoredEvent[]; -}; +type AgentDetailData = { agent: Agent; checks: CheckState[] }; +type AgentPollPayload = AgentDetailData & { events?: StoredEvent[] }; const severityRank: Record = { critical: 0, @@ -33,22 +42,24 @@ const severityRank: Record = { export function AgentDetailBrowser({ initialData, + initialEvents, sort, tab, + eventsLimit, + eventsCursor, }: { initialData: AgentDetailData; + initialEvents: EventsPageData; sort: SortKey; tab: TabKey; + eventsLimit: PageSize; + eventsCursor?: string; }) { - const { data } = usePolling( - initialData, - `/api/poll/agent?agent_id=${encodeURIComponent(initialData.agent.agent_id)}`, - ); - const { agent, events } = data; - const checks = useMemo( - () => sortChecks(withLivenessChecks([agent], data.checks), sort), - [agent, data.checks, sort], - ); + const agentId = initialData.agent.agent_id; + const pollUrl = `/api/poll/agent?agent_id=${encodeURIComponent(agentId)}`; + const { data } = usePolling(initialData, pollUrl); + const agent = data.agent; + const checks = useMemo(() => sortChecks(data.checks, sort), [data.checks, sort]); return (
@@ -76,28 +87,32 @@ export function AgentDetailBrowser({ - - {tab === "checks" ? : } -
- ); -} - -function TabNav({ agentId, current, sort }: { agentId: string; current: TabKey; sort: SortKey }) { - const base = `/agents/${encodeURIComponent(agentId)}`; - return ( -
- - Checks - - - Events - + + {tab === "checks" ? ( + + ) : ( + + )}
); } @@ -111,9 +126,22 @@ function ChecksTable({ checks: CheckState[]; sort: SortKey; }) { - if (checks.length === 0) return
No checks.
; + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; + const pageCount = pageCountOf(checks.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(checks, currentPage, pageSize); + + if (checks.length === 0) return ; return (
+
+ +
@@ -136,7 +164,7 @@ function ChecksTable({ - {checks.map((c) => ( + {visible.map((c) => ( - + ))}
@@ -145,43 +173,81 @@ function ChecksTable({ {c.summary ?? "—"}
+
); } -function EventsTable({ events }: { events: StoredEvent[] }) { - if (events.length === 0) return
No events.
; +function EventsTable({ + agentId, + sort, + eventsLimit, + eventsCursor, + initialEvents, +}: { + agentId: string; + sort: SortKey; + eventsLimit: PageSize; + eventsCursor?: string; + initialEvents: EventsPageData; +}) { + const router = useRouter(); + const pollUrl = useMemo(() => { + const qs = new URLSearchParams({ agent_id: agentId, limit: String(eventsLimit) }); + if (eventsCursor) qs.set("cursor", eventsCursor); + return `/api/poll/events?${qs.toString()}`; + }, [agentId, eventsCursor, eventsLimit]); + const { data } = usePolling(initialEvents, pollUrl); + + const onLimitChange = (next: PageSize) => { + const qs = new URLSearchParams({ tab: "events", sort, events_limit: String(next) }); + router.push(`/agents/${encodeURIComponent(agentId)}?${qs.toString()}`); + }; + return (
- - - - - - - - - - - {events.map((e) => ( - - - - - +
+ +
+ {data.items.length === 0 ? ( + + ) : ( +
observed_atcheck_idstatusduration_ms
- - - - {e.status}{e.duration_ms}
+ + + + + + - ))} - -
observed_atcheck_idstatusduration_ms
+ + + {data.items.map((e) => ( + + + + + + + + {e.status} + {e.duration_ms} + + ))} + + + )} +
); } diff --git a/web/src/components/AgentLink.tsx b/web/src/components/AgentLink.tsx new file mode 100644 index 0000000..c4aca1d --- /dev/null +++ b/web/src/components/AgentLink.tsx @@ -0,0 +1,19 @@ +import Link from "next/link"; + +import type { components } from "@/lib/api-types"; + +type Agent = components["schemas"]["Agent"]; + +export function AgentLink({ agent, agentId }: { agent?: Agent; agentId: string }) { + const hostname = agent?.hostname || agentId; + const showId = agent && agent.hostname && agent.hostname !== agentId; + return ( + + {hostname} + {showId ? {agentId} : null} + + ); +} diff --git a/web/src/components/AgentsBrowser.tsx b/web/src/components/AgentsBrowser.tsx index a01be20..d615a1c 100644 --- a/web/src/components/AgentsBrowser.tsx +++ b/web/src/components/AgentsBrowser.tsx @@ -1,13 +1,24 @@ "use client"; -import Link from "next/link"; import { useMemo, useState } from "react"; import { AgentFeatures } from "@/components/AgentFeatures"; +import { AgentLink } from "@/components/AgentLink"; +import { ClientPager } from "@/components/ClientPager"; +import { EmptyPanel, Td, Th } from "@/components/DataPanel"; import { LabelChips } from "@/components/Labels"; +import { ListToolbar } from "@/components/ListToolbar"; +import { SortHeaderButton, type SortDir } from "@/components/SortHeaderButton"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; import { statusBadgeClass } from "@/lib/format"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; import type { InventoryData } from "@/lib/view-types"; @@ -16,7 +27,6 @@ type CheckState = components["schemas"]["CheckState"]; type CheckStatus = CheckState["status"]; type CheckCounts = Record; type SortField = "host" | "status" | "last_seen_at"; -type SortDir = "asc" | "desc"; export type AgentRow = { agent: Agent; checks: CheckCounts }; @@ -36,77 +46,75 @@ export function AgentsBrowser({ initialData }: { initialData: InventoryData }) { [dir, initialRows, query, sort], ); + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const pageCount = pageCountOf(rows.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(rows, currentPage, pageSize); + + const onQueryChange = (value: string) => { + setQuery(value); + setPage(1); + }; + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; const chooseSort = (field: SortField) => { if (field === sort) { setDir((current) => (current === "asc" ? "desc" : "asc")); - return; + } else { + setSort(field); + setDir(fieldDefaultDir(field)); } - setSort(field); - setDir(fieldDefaultDir(field)); + setPage(1); }; return ( <>

Agents

- {rows.length} items -
-
- setQuery(event.target.value)} - placeholder="search agent" - className="min-w-0 flex-1 border border-neutral-800 bg-neutral-950 px-2 py-1.5 text-sm text-neutral-100 outline-none focus:border-neutral-500" - /> - {query ? ( - - ) : null} + + {rows.length} of {initialRows.length} items +
+ {rows.length === 0 ? ( -
No data.
+ ) : ( - {rows.map(({ agent: a, checks }) => ( + {visible.map(({ agent: a, checks }) => (
- + HOST - + - + STATUS - + - + LAST_SEEN_AT - + features labels
- - {a.hostname} - {a.agent_id !== a.hostname ? ( - {a.agent_id} - ) : null} - + @@ -125,50 +133,11 @@ export function AgentsBrowser({ initialData }: { initialData: InventoryData }) {
)} + ); } -function Th({ children }: { children: React.ReactNode }) { - return ( - - {children} - - ); -} - -function Td({ children, className = "" }: { children: React.ReactNode; className?: string }) { - return ( - {children} - ); -} - -function SortButton({ - field, - current, - dir, - onClick, - children, -}: { - field: SortField; - current: SortField; - dir: SortDir; - onClick: (field: SortField) => void; - children: React.ReactNode; -}) { - const active = field === current; - return ( - - ); -} - function filterAgents(rows: AgentRow[], query: string): AgentRow[] { const needle = query.trim().toLowerCase(); if (!needle) return rows; diff --git a/web/src/components/CheckDetailBrowser.tsx b/web/src/components/CheckDetailBrowser.tsx index 1b6222a..8d19aab 100644 --- a/web/src/components/CheckDetailBrowser.tsx +++ b/web/src/components/CheckDetailBrowser.tsx @@ -1,20 +1,27 @@ "use client"; import Link from "next/link"; -import { useMemo } from "react"; +import { useRouter } from "next/navigation"; +import { useMemo, useState } from "react"; +import { AgentLink } from "@/components/AgentLink"; import { CheckStatusSummary } from "@/components/CheckStatusSummary"; -import { Td, Th } from "@/components/DataPanel"; +import { ClientPager } from "@/components/ClientPager"; +import { EmptyPanel, SummaryCell, Td, Th } from "@/components/DataPanel"; +import { DetailTabs } from "@/components/DetailTabs"; +import { PageSizeSelect } from "@/components/PageSizeSelect"; +import { Pager } from "@/components/Pager"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; -import { - groupChecks, - statusSeverity, - withLivenessChecks, - type Agent, - type CheckGroup, -} from "@/lib/check-groups"; +import { groupChecks, statusSeverity, type Agent, type CheckGroup } from "@/lib/check-groups"; import { statusBadgeClass } from "@/lib/format"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; import type { InventoryData } from "@/lib/view-types"; @@ -27,21 +34,24 @@ export function CheckDetailBrowser({ initialData, initialEvents, tab, + eventsLimit, + eventsCursor, }: { checkId: string; initialData: InventoryData; initialEvents: EventsPageData; tab: TabKey; + eventsLimit: PageSize; + eventsCursor?: string; }) { const { data } = usePolling(initialData, "/api/poll/inventory"); - const agentsById = useMemo(() => new Map(data.agents.map((agent) => [agent.agent_id, agent])), [data.agents]); - const checks = useMemo( - () => withLivenessChecks(data.agents, data.checks), - [data.agents, data.checks], + const agentsById = useMemo( + () => new Map(data.agents.map((agent) => [agent.agent_id, agent])), + [data.agents], ); const group = useMemo( - () => groupChecks(checks, agentsById).find((item) => item.checkId === checkId), - [agentsById, checkId, checks], + () => groupChecks(data.checks, agentsById).find((item) => item.checkId === checkId), + [agentsById, checkId, data.checks], ); if (!group) { @@ -76,76 +86,88 @@ export function CheckDetailBrowser({ - + {tab === "agents" ? ( ) : ( - + )} ); } -function TabNav({ checkId, current }: { checkId: string; current: TabKey }) { - const base = `/checks/${encodeURIComponent(checkId)}`; - return ( -
- - Agents - - - Events - -
- ); -} - function AgentsForCheck({ group }: { group: CheckGroup }) { - const rows = [...group.checks].sort( - (a, b) => - statusSeverity[b.check.status] - statusSeverity[a.check.status] || - hostName(a).localeCompare(hostName(b)), + const rows = useMemo( + () => + [...group.checks].sort( + (a, b) => + statusSeverity[b.check.status] - statusSeverity[a.check.status] || + hostName(a).localeCompare(hostName(b)), + ), + [group.checks], ); + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; + const pageCount = pageCountOf(rows.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(rows, currentPage, pageSize); + return ( - - - - - - - - - - - {rows.map(({ check, agent }) => ( - - - - - +
+
+ +
+
HOSTSTATUSLAST_OBSERVED_ATSUMMARY
- - {agent?.hostname || check.agent_id} - {agent && agent.hostname !== check.agent_id ? ( - {check.agent_id} - ) : null} - - {check.status} - - {check.summary ?? "—"}
+ + + + + + - ))} - -
HOSTSTATUSLAST_OBSERVED_ATSUMMARY
+ + + {visible.map(({ check, agent }) => ( + + + + + {check.status} + + + + + + ))} + + + + ); } @@ -153,57 +175,81 @@ function EventsForCheck({ checkId, initialData, agentsById, + eventsLimit, + eventsCursor, }: { checkId: string; initialData: EventsPageData; agentsById: Map; + eventsLimit: PageSize; + eventsCursor?: string; }) { - const { data } = usePolling( - initialData, - `/api/poll/events?check_id=${encodeURIComponent(checkId)}&limit=50`, - ); + const router = useRouter(); + const pollUrl = useMemo(() => { + const qs = new URLSearchParams({ + check_id: checkId, + limit: String(eventsLimit), + }); + if (eventsCursor) qs.set("cursor", eventsCursor); + return `/api/poll/events?${qs.toString()}`; + }, [checkId, eventsCursor, eventsLimit]); + const { data } = usePolling(initialData, pollUrl); - if (data.items.length === 0) return
No events.
; + const onLimitChange = (next: PageSize) => { + router.push( + `/checks/${encodeURIComponent(checkId)}?tab=events&events_limit=${next}`, + ); + }; return ( - - - - - - - - - - - - - {data.items.map((event) => { - const agent = agentsById.get(event.agent_id); - return ( - - - - - - - +
+
+ +
+ {data.items.length === 0 ? ( + + ) : ( +
OBSERVED_ATRECEIVED_ATHOSTSTATUSDURATION_MSSUMMARY
- - - - - - {agent?.hostname || event.agent_id} - - {event.status}{event.duration_ms}{event.output ?? "—"}
+ + + + + + + + - ); - })} - -
OBSERVED_ATRECEIVED_ATHOSTSTATUSDURATION_MSSUMMARY
+ + + {data.items.map((event) => { + const agent = agentsById.get(event.agent_id); + return ( + + + + + + + + + + + {event.status} + {event.duration_ms} + + + ); + })} + + + )} + + ); } diff --git a/web/src/components/ChecksBrowser.tsx b/web/src/components/ChecksBrowser.tsx index 3a36fca..ae3a6ce 100644 --- a/web/src/components/ChecksBrowser.tsx +++ b/web/src/components/ChecksBrowser.tsx @@ -4,17 +4,27 @@ import Link from "next/link"; import { useMemo, useState } from "react"; import { CheckStatusSummary } from "@/components/CheckStatusSummary"; +import { ClientPager } from "@/components/ClientPager"; +import { EmptyPanel, Td, Th } from "@/components/DataPanel"; +import { ListToolbar } from "@/components/ListToolbar"; +import { SortHeaderButton, type SortDir } from "@/components/SortHeaderButton"; import { DateTime } from "@/components/TimeZoneControls"; import { compareStatusCounts, groupChecks, type CheckGroup, } from "@/lib/check-groups"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; import type { InventoryData } from "@/lib/view-types"; type SortField = "check" | "status" | "last_observed_at"; -type SortDir = "asc" | "desc"; export function ChecksBrowser({ initialData }: { initialData: InventoryData }) { const { data } = usePolling(initialData, "/api/poll/inventory"); @@ -29,65 +39,71 @@ export function ChecksBrowser({ initialData }: { initialData: InventoryData }) { [dir, groups, query, sort], ); + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const pageCount = pageCountOf(rows.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(rows, currentPage, pageSize); + + const onQueryChange = (value: string) => { + setQuery(value); + setPage(1); + }; + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; const chooseSort = (field: SortField) => { if (field === sort) { setDir((current) => (current === "asc" ? "desc" : "asc")); - return; + } else { + setSort(field); + setDir(fieldDefaultDir(field)); } - setSort(field); - setDir(fieldDefaultDir(field)); + setPage(1); }; return ( <>

Checks

- {rows.length} items -
-
- setQuery(event.target.value)} - placeholder="search check" - className="min-w-0 flex-1 border border-neutral-800 bg-neutral-950 px-2 py-1.5 text-sm text-neutral-100 outline-none focus:border-neutral-500" - /> - {query ? ( - - ) : null} + + {rows.length} of {groups.length} items +
+ {rows.length === 0 ? ( -
No data.
+ ) : ( - {rows.map((row) => ( + {visible.map((row) => (
- + CHECK - + - + STATUS - + - + LAST_OBSERVED_AT - + AGENTS
)} + ); } -function Th({ children }: { children: React.ReactNode }) { - return ( - - {children} - - ); -} - -function Td({ children, className = "" }: { children: React.ReactNode; className?: string }) { - return {children}; -} - -function SortButton({ - field, - current, - dir, - onClick, - children, -}: { - field: SortField; - current: SortField; - dir: SortDir; - onClick: (field: SortField) => void; - children: React.ReactNode; -}) { - const active = field === current; - return ( - - ); -} - function filterChecks(rows: CheckGroup[], query: string): CheckGroup[] { const needle = query.trim().toLowerCase(); if (!needle) return rows; diff --git a/web/src/components/ClientPager.tsx b/web/src/components/ClientPager.tsx new file mode 100644 index 0000000..4529841 --- /dev/null +++ b/web/src/components/ClientPager.tsx @@ -0,0 +1,39 @@ +"use client"; + +export function ClientPager({ + page, + pageCount, + onPageChange, +}: { + page: number; + pageCount: number; + onPageChange: (page: number) => void; +}) { + const prev = () => onPageChange(Math.max(1, page - 1)); + const next = () => onPageChange(Math.min(pageCount, page + 1)); + const disabledPrev = page <= 1; + const disabledNext = page >= pageCount; + return ( +
+ + + page {page} of {pageCount} + + +
+ ); +} diff --git a/web/src/components/DataPanel.tsx b/web/src/components/DataPanel.tsx index 7d901bd..f7c1010 100644 --- a/web/src/components/DataPanel.tsx +++ b/web/src/components/DataPanel.tsx @@ -39,3 +39,7 @@ export function Td({ children, className = "" }: { children: React.ReactNode; cl {children} ); } + +export function SummaryCell({ text }: { text?: string | null }) { + return {text ?? "—"}; +} diff --git a/web/src/components/DetailTabs.tsx b/web/src/components/DetailTabs.tsx new file mode 100644 index 0000000..1f5566c --- /dev/null +++ b/web/src/components/DetailTabs.tsx @@ -0,0 +1,23 @@ +import Link from "next/link"; + +export type DetailTab = { key: string; label: string; href: string }; + +export function DetailTabs({ tabs, current }: { tabs: DetailTab[]; current: string }) { + return ( +
+ {tabs.map((tab) => ( + + {tab.label} + + ))} +
+ ); +} diff --git a/web/src/components/EventsBrowser.tsx b/web/src/components/EventsBrowser.tsx index 0631ea6..be5278c 100644 --- a/web/src/components/EventsBrowser.tsx +++ b/web/src/components/EventsBrowser.tsx @@ -1,14 +1,17 @@ "use client"; import Link from "next/link"; +import { useRouter } from "next/navigation"; import { useMemo } from "react"; import { CheckEventsLink } from "@/components/CheckEventsLink"; -import { EmptyPanel, PageHeader, Td, Th } from "@/components/DataPanel"; +import { EmptyPanel, PageHeader, SummaryCell, Td, Th } from "@/components/DataPanel"; +import { PageSizeSelect } from "@/components/PageSizeSelect"; import { Pager } from "@/components/Pager"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; import { statusBadgeClass } from "@/lib/format"; +import { type PageSize } from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; type StoredEvent = components["schemas"]["StoredCheckResultEvent"]; @@ -19,19 +22,32 @@ export function EventsBrowser({ cursor, agentId, checkId, + limit, }: { initialData: EventsPageData; cursor?: string; agentId?: string; checkId?: string; + limit: PageSize; }) { - const url = useMemo(() => eventsPollUrl({ cursor, agentId, checkId }), [agentId, checkId, cursor]); + const router = useRouter(); + const url = useMemo( + () => eventsPollUrl({ cursor, agentId, checkId, limit }), + [agentId, checkId, cursor, limit], + ); const { data } = usePolling(initialData, url); + const onLimitChange = (next: PageSize) => { + router.push(eventsHref({ agent_id: agentId, check_id: checkId, limit: next })); + }; + return ( <> - +
+ + +
{data.items.length === 0 ? ( ) : ( @@ -62,7 +78,7 @@ export function EventsBrowser({ {e.status} {e.duration_ms} - {e.output ?? "—"} + ))} @@ -72,24 +88,43 @@ export function EventsBrowser({ basePath="/events" cursor={cursor} nextCursor={data.next_cursor} - extra={{ agent_id: agentId, check_id: checkId }} + extra={{ agent_id: agentId, check_id: checkId, limit: String(limit) }} /> ); } -function ActiveFilters({ agentId, checkId }: { agentId?: string; checkId?: string }) { - if (!agentId && !checkId) return null; +function ActiveFilters({ + agentId, + checkId, + limit, +}: { + agentId?: string; + checkId?: string; + limit: PageSize; +}) { + if (!agentId && !checkId) return
; return ( -
+
{agentId ? ( - + ) : null} {checkId ? ( - + ) : null} - + clear
@@ -110,10 +145,15 @@ function FilterChip({ label, value, href }: { label: string; value: string; href ); } -function eventsHref(params: { agent_id?: string; check_id?: string }): string { +function eventsHref(params: { + agent_id?: string; + check_id?: string; + limit?: PageSize; +}): string { const qs = new URLSearchParams(); if (params.agent_id) qs.set("agent_id", params.agent_id); if (params.check_id) qs.set("check_id", params.check_id); + if (params.limit) qs.set("limit", String(params.limit)); const query = qs.toString(); return query ? `/events?${query}` : "/events"; } @@ -122,15 +162,17 @@ function eventsPollUrl({ cursor, agentId, checkId, + limit, }: { cursor?: string; agentId?: string; checkId?: string; + limit: PageSize; }) { const qs = new URLSearchParams(); if (cursor) qs.set("cursor", cursor); if (agentId) qs.set("agent_id", agentId); if (checkId) qs.set("check_id", checkId); - const query = qs.toString(); - return query ? `/api/poll/events?${query}` : "/api/poll/events"; + qs.set("limit", String(limit)); + return `/api/poll/events?${qs.toString()}`; } diff --git a/web/src/components/IncidentsBrowser.tsx b/web/src/components/IncidentsBrowser.tsx index dc157d3..e91b97f 100644 --- a/web/src/components/IncidentsBrowser.tsx +++ b/web/src/components/IncidentsBrowser.tsx @@ -3,11 +3,22 @@ import Link from "next/link"; import { useMemo, useState } from "react"; +import { AgentLink } from "@/components/AgentLink"; import { CheckEventsLink } from "@/components/CheckEventsLink"; -import { EmptyPanel, PageHeader, Td, Th } from "@/components/DataPanel"; +import { ClientPager } from "@/components/ClientPager"; +import { EmptyPanel, PageHeader, SummaryCell, Td, Th } from "@/components/DataPanel"; +import { PageSizeSelect } from "@/components/PageSizeSelect"; +import { SortHeaderButton, type SortDir } from "@/components/SortHeaderButton"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; import { statusBadgeClass } from "@/lib/format"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; import type { InventoryData } from "@/lib/view-types"; @@ -15,104 +26,130 @@ const STATES = ["all", "open", "resolved"] as const; type Incident = components["schemas"]["Incident"]; type Agent = components["schemas"]["Agent"]; -type IncidentsPageData = { items: Incident[] }; +type IncidentsData = { items: Incident[] }; type SortField = "status" | "host" | "check" | "opened_at" | "resolved_at"; -type SortDir = "asc" | "desc"; -const stateRank: Record = { resolved: 0, open: 1 }; -const severityRank: Record = { unknown: 1, warning: 2, critical: 3 }; +const stateRank: Record = { open: 0, resolved: 1 }; +const severityRank: Record = { + critical: 0, + warning: 1, + unknown: 2, +}; export function IncidentsBrowser({ - initialData, + initialItems, initialInventory, state, }: { - initialData: IncidentsPageData; + initialItems: Incident[]; initialInventory: InventoryData; state?: "open" | "resolved"; }) { - const url = useMemo(() => incidentsPollUrl({ state }), [state]); - const { data } = usePolling(initialData, url); + const url = useMemo( + () => (state ? `/api/poll/incidents?state=${state}` : "/api/poll/incidents"), + [state], + ); + const { data } = usePolling({ items: initialItems }, url); const { data: inventory } = usePolling(initialInventory, "/api/poll/inventory"); - const [sort, setSort] = useState("status"); - const [dir, setDir] = useState("desc"); const agentsById = useMemo( () => new Map(inventory.agents.map((agent) => [agent.agent_id, agent])), [inventory.agents], ); - const rows = useMemo( + + const [sort, setSort] = useState("status"); + const [dir, setDir] = useState("asc"); + const sorted = useMemo( () => sortIncidents(data.items, agentsById, sort, dir), [agentsById, data.items, dir, sort], ); + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; const chooseSort = (field: SortField) => { if (field === sort) { setDir((current) => (current === "asc" ? "desc" : "asc")); - return; + } else { + setSort(field); + setDir(fieldDefaultDir(field)); } - setSort(field); - setDir(fieldDefaultDir(field)); + setPage(1); }; + const pageCount = pageCountOf(sorted.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(sorted, currentPage, pageSize); + return ( <> - -
- {STATES.map((s) => { - const href = s === "all" ? "/incidents" : `/incidents?state=${s}`; - const active = (s === "all" && !state) || s === state; - return ( - - {s} - - ); - })} + +
+
+ {STATES.map((s) => { + const href = s === "all" ? "/incidents" : `/incidents?state=${s}`; + const active = (s === "all" && !state) || s === state; + return ( + + {s} + + ); + })} +
+
- {rows.length === 0 ? ( + {sorted.length === 0 ? ( ) : ( - {rows.map(({ incident, agent }) => ( - + {visible.map((incident) => ( + ))}
- + STATUS - + - + HOST - + - + CHECK - + - + OPENED_AT - + - + RESOLVED_AT - + SUMMARY
)} + ); } @@ -128,21 +165,7 @@ function IncidentRow({ incident, agent }: { incident: Incident; agent?: Agent }) {incident.severity}
- - {agentId ? ( - - {agent?.hostname || agentId} - {agent && agent.hostname !== agentId ? ( - {agentId} - ) : null} - - ) : ( - "—" - )} - + {agentId ? : "—"} @@ -152,7 +175,7 @@ function IncidentRow({ incident, agent }: { incident: Incident; agent?: Agent }) - {incident.summary ?? "—"} + ); } @@ -172,97 +195,6 @@ function IncidentCheckLink({ agentId, checkId }: { agentId: string; checkId: str return agentId ? : {checkId}; } -function SortButton({ - field, - current, - dir, - onClick, - children, -}: { - field: SortField; - current: SortField; - dir: SortDir; - onClick: (field: SortField) => void; - children: React.ReactNode; -}) { - const active = field === current; - return ( - - ); -} - -function fieldDefaultDir(field: SortField): SortDir { - return field === "host" || field === "check" ? "asc" : "desc"; -} - -function sortIncidents( - incidents: Incident[], - agentsById: Map, - sort: SortField, - dir: SortDir, -): { incident: Incident; agent?: Agent }[] { - return incidents - .map((incident) => ({ incident, agent: agentsById.get(incidentAgentId(incident)) })) - .sort((a, b) => { - const primary = compareByField(a, b, sort); - if (primary !== 0) return dir === "asc" ? primary : -primary; - return fallbackCompare(a, b); - }); -} - -function compareByField( - a: { incident: Incident; agent?: Agent }, - b: { incident: Incident; agent?: Agent }, - sort: SortField, -): number { - switch (sort) { - case "host": - return hostName(a).localeCompare(hostName(b)) || incidentAgentId(a.incident).localeCompare(incidentAgentId(b.incident)); - case "check": - return incidentCheckId(a.incident).localeCompare(incidentCheckId(b.incident)); - case "opened_at": - return Date.parse(a.incident.opened_at) - Date.parse(b.incident.opened_at); - case "resolved_at": - return dateValue(a.incident.resolved_at) - dateValue(b.incident.resolved_at); - case "status": - return compareIncidentImportance(a.incident, b.incident); - } -} - -function compareIncidentImportance(a: Incident, b: Incident): number { - return ( - stateRank[a.state] - stateRank[b.state] || - severityRank[a.severity] - severityRank[b.severity] || - Date.parse(a.opened_at) - Date.parse(b.opened_at) - ); -} - -function fallbackCompare( - a: { incident: Incident; agent?: Agent }, - b: { incident: Incident; agent?: Agent }, -): number { - return ( - compareIncidentImportance(a.incident, b.incident) || - hostName(a).localeCompare(hostName(b)) || - incidentCheckId(a.incident).localeCompare(incidentCheckId(b.incident)) - ); -} - -function dateValue(value?: string | null): number { - return value ? Date.parse(value) : 0; -} - -function hostName(row: { incident: Incident; agent?: Agent }): string { - return row.agent?.hostname || incidentAgentId(row.incident); -} - function incidentAgentId(incident: Incident): string { return incident.agent_id ?? ""; } @@ -271,6 +203,56 @@ function incidentCheckId(incident: Incident): string { return incident.check_id ?? ""; } -function incidentsPollUrl({ state }: { state?: string }) { - return state ? `/api/poll/incidents?state=${encodeURIComponent(state)}` : "/api/poll/incidents"; +function fieldDefaultDir(field: SortField): SortDir { + switch (field) { + case "status": + return "asc"; + case "host": + case "check": + return "asc"; + case "opened_at": + case "resolved_at": + return "desc"; + } +} + +function sortIncidents( + items: Incident[], + agentsById: Map, + sort: SortField, + dir: SortDir, +): Incident[] { + return [...items].sort((a, b) => { + const primary = compareIncidents(a, b, agentsById, sort); + if (primary !== 0) return dir === "asc" ? primary : -primary; + return (b.opened_at ?? "").localeCompare(a.opened_at ?? ""); + }); +} + +function compareIncidents( + a: Incident, + b: Incident, + agentsById: Map, + sort: SortField, +): number { + switch (sort) { + case "status": { + const s = stateRank[a.state] - stateRank[b.state]; + if (s !== 0) return s; + return severityRank[a.severity] - severityRank[b.severity]; + } + case "host": + return hostFor(a, agentsById).localeCompare(hostFor(b, agentsById)); + case "check": + return incidentCheckId(a).localeCompare(incidentCheckId(b)); + case "opened_at": + return (a.opened_at ?? "").localeCompare(b.opened_at ?? ""); + case "resolved_at": + return (a.resolved_at ?? "").localeCompare(b.resolved_at ?? ""); + } +} + +function hostFor(incident: Incident, agentsById: Map): string { + const id = incidentAgentId(incident); + return agentsById.get(id)?.hostname || id; } diff --git a/web/src/components/ListToolbar.tsx b/web/src/components/ListToolbar.tsx new file mode 100644 index 0000000..d7816f0 --- /dev/null +++ b/web/src/components/ListToolbar.tsx @@ -0,0 +1,42 @@ +"use client"; + +import { PageSizeSelect } from "@/components/PageSizeSelect"; +import { type PageSize } from "@/lib/pagination"; + +export function ListToolbar({ + query, + onQueryChange, + placeholder, + pageSize, + onPageSizeChange, +}: { + query: string; + onQueryChange: (value: string) => void; + placeholder: string; + pageSize: PageSize; + onPageSizeChange: (next: PageSize) => void; +}) { + return ( +
+
+ onQueryChange(event.target.value)} + placeholder={placeholder} + className="min-w-0 flex-1 border border-neutral-800 bg-neutral-950 px-2 py-1.5 text-sm text-neutral-100 outline-none focus:border-neutral-500" + /> + {query ? ( + + ) : null} +
+ +
+ ); +} diff --git a/web/src/components/OutboxBrowser.tsx b/web/src/components/OutboxBrowser.tsx index d99e1ee..09cd4fb 100644 --- a/web/src/components/OutboxBrowser.tsx +++ b/web/src/components/OutboxBrowser.tsx @@ -1,33 +1,45 @@ "use client"; -import { useMemo } from "react"; +import { useState } from "react"; +import { ClientPager } from "@/components/ClientPager"; import { EmptyPanel, PageHeader, Td, Th } from "@/components/DataPanel"; -import { Pager } from "@/components/Pager"; +import { PageSizeSelect } from "@/components/PageSizeSelect"; import { DateTime } from "@/components/TimeZoneControls"; import type { components } from "@/lib/api-types"; import { statusBadgeClass } from "@/lib/format"; +import { + DEFAULT_PAGE_SIZE, + clampPage, + pageCountOf, + paginate, + type PageSize, +} from "@/lib/pagination"; import { usePolling } from "@/lib/usePolling"; type OutboxItem = components["schemas"]["NotificationOutboxItem"]; -type OutboxPageData = { items: OutboxItem[]; next_cursor?: string | null }; +type OutboxData = { items: OutboxItem[] }; -export function OutboxBrowser({ - initialData, - cursor, -}: { - initialData: OutboxPageData; - cursor?: string; -}) { - const url = useMemo(() => { - if (!cursor) return "/api/poll/outbox"; - return `/api/poll/outbox?cursor=${encodeURIComponent(cursor)}`; - }, [cursor]); - const { data } = usePolling(initialData, url); +export function OutboxBrowser({ initialItems }: { initialItems: OutboxItem[] }) { + const { data } = usePolling({ items: initialItems }, "/api/poll/outbox"); + + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [page, setPage] = useState(1); + const onPageSizeChange = (next: PageSize) => { + setPageSize(next); + setPage(1); + }; + + const pageCount = pageCountOf(data.items.length, pageSize); + const currentPage = clampPage(page, pageCount); + const visible = paginate(data.items, currentPage, pageSize); return ( <> +
+ +
{data.items.length === 0 ? ( ) : ( @@ -44,7 +56,7 @@ export function OutboxBrowser({ - {data.items.map((o) => ( + {visible.map((o) => ( @@ -62,7 +74,7 @@ export function OutboxBrowser({ )} - + ); } diff --git a/web/src/components/PageSizeSelect.tsx b/web/src/components/PageSizeSelect.tsx new file mode 100644 index 0000000..b691be7 --- /dev/null +++ b/web/src/components/PageSizeSelect.tsx @@ -0,0 +1,30 @@ +"use client"; + +import { PAGE_SIZE_OPTIONS, type PageSize } from "@/lib/pagination"; + +export function PageSizeSelect({ + value, + onChange, + options = PAGE_SIZE_OPTIONS, +}: { + value: PageSize; + onChange: (size: PageSize) => void; + options?: readonly number[]; +}) { + return ( + + ); +} diff --git a/web/src/components/SortHeaderButton.tsx b/web/src/components/SortHeaderButton.tsx new file mode 100644 index 0000000..cc38fea --- /dev/null +++ b/web/src/components/SortHeaderButton.tsx @@ -0,0 +1,31 @@ +"use client"; + +import type { ReactNode } from "react"; + +export type SortDir = "asc" | "desc"; + +export function SortHeaderButton({ + field, + current, + dir, + onClick, + children, +}: { + field: F; + current: F; + dir: SortDir; + onClick: (field: F) => void; + children: ReactNode; +}) { + const active = field === current; + return ( + + ); +} diff --git a/web/src/lib/check-groups.ts b/web/src/lib/check-groups.ts index 92d09ae..3170e3f 100644 --- a/web/src/lib/check-groups.ts +++ b/web/src/lib/check-groups.ts @@ -2,7 +2,6 @@ import type { components } from "@/lib/api-types"; export type Agent = components["schemas"]["Agent"]; export type CheckState = components["schemas"]["CheckState"]; -export type StoredEvent = components["schemas"]["StoredCheckResultEvent"]; export type CheckStatus = CheckState["status"]; export type CheckCounts = Record; @@ -44,67 +43,6 @@ export function groupChecks(checks: CheckState[], agentsById: Map return [...byCheck.values()]; } -export function withLivenessChecks(agents: Agent[], checks: CheckState[]): CheckState[] { - const regularChecks = checks.filter((check) => check.check_id !== LIVENESS_CHECK_ID); - const livenessChecks = agents - .map(livenessCheckForAgent) - .filter((check): check is CheckState => check !== null); - return [...regularChecks, ...livenessChecks]; -} - -export function livenessCheckForAgent(agent: Agent): CheckState | null { - if (agent.status !== "stale" && agent.status !== "dead") return null; - const status: CheckStatus = agent.status === "dead" ? "critical" : "warning"; - return { - agent_id: agent.agent_id, - check_id: LIVENESS_CHECK_ID, - status, - last_observed_at: agent.last_seen_at, - exit_code: status === "critical" ? 2 : 1, - incident_key: `agent:${agent.agent_id}:liveness`, - summary: `agent is ${agent.status}`, - }; -} - -export function withLivenessEvents( - agents: Agent[], - events: StoredEvent[], - filters: { agentId?: string; checkId?: string; cursor?: string } = {}, -): StoredEvent[] { - if (filters.cursor) return events; - const livenessEvents = agents - .map(livenessEventForAgent) - .filter((event): event is StoredEvent => event !== null) - .filter((event) => { - if (filters.agentId && event.agent_id !== filters.agentId) return false; - if (filters.checkId && event.check_id !== filters.checkId) return false; - return true; - }); - return [...events, ...livenessEvents].sort((a, b) => b.observed_at.localeCompare(a.observed_at)); -} - -export function livenessEventForAgent(agent: Agent): StoredEvent | null { - const check = livenessCheckForAgent(agent); - if (!check) return null; - // Synthetic event represents the current liveness state, not the last heartbeat, - // so observed_at uses "now" to keep it at the top of the events list across polls. - const now = new Date().toISOString(); - return { - event_id: `synthetic:${LIVENESS_CHECK_ID}:${agent.agent_id}:${agent.status}:${agent.last_seen_at}`, - agent_id: agent.agent_id, - check_id: LIVENESS_CHECK_ID, - observed_at: now, - received_at: now, - status: check.status, - exit_code: check.status === "critical" ? 2 : 1, - duration_ms: 0, - output: check.summary, - output_truncated: false, - notifications_enabled: false, - incident_key: check.incident_key, - }; -} - export function emptyCheckCounts(): CheckCounts { return { ok: 0, warning: 0, critical: 0, unknown: 0 }; } diff --git a/web/src/lib/loaders.ts b/web/src/lib/loaders.ts index a6c4291..8d028db 100644 --- a/web/src/lib/loaders.ts +++ b/web/src/lib/loaders.ts @@ -1,8 +1,7 @@ import "server-only"; -import { ApiError, api, type Agent, type CheckState, type StoredEvent } from "@/lib/api"; +import { api, type Agent, type CheckState, type OutboxItem, type StoredEvent } from "@/lib/api"; import type { components } from "@/lib/api-types"; -import { LIVENESS_CHECK_ID, withLivenessChecks, withLivenessEvents } from "@/lib/check-groups"; import type { InventoryData, OverviewData } from "@/lib/view-types"; type Incident = components["schemas"]["Incident"]; @@ -31,10 +30,10 @@ export async function loadChecks(agentId?: string): Promise { export async function loadInventory(): Promise { const [agents, checks] = await Promise.all([loadAgents(), loadChecks()]); - return { agents, checks: withLivenessChecks(agents, checks) }; + return { agents, checks }; } -export async function loadEventsWithLiveness({ +export async function loadEvents({ agentId, checkId, cursor, @@ -45,28 +44,10 @@ export async function loadEventsWithLiveness({ cursor?: string; limit?: number; } = {}): Promise<{ items: StoredEvent[]; next_cursor?: string | null }> { - const data = await api.queryEvents({ agent_id: agentId, check_id: checkId, cursor, limit }); - if (cursor || (checkId && checkId !== LIVENESS_CHECK_ID)) return data; - - const agents = agentId ? await loadOptionalAgent(agentId) : await loadAgents(); - return { - ...data, - items: withLivenessEvents(agents, data.items, { agentId, checkId, cursor }), - }; + return api.queryEvents({ agent_id: agentId, check_id: checkId, cursor, limit }); } -async function loadOptionalAgent(agentId: string): Promise { - try { - return [await api.getAgent(agentId)]; - } catch (e) { - if (e instanceof ApiError && e.status === 404) return []; - throw e; - } -} - -export async function loadIncidents( - state?: "open" | "resolved", -): Promise<{ items: Incident[] }> { +export async function loadIncidents(state?: "open" | "resolved"): Promise { const items: Incident[] = []; let cursor: string | undefined; do { @@ -74,25 +55,46 @@ export async function loadIncidents( items.push(...page.items); cursor = page.next_cursor ?? undefined; } while (cursor); - return { items }; + return items; +} + +export async function loadOutboxAll(): Promise { + const items: OutboxItem[] = []; + let cursor: string | undefined; + do { + const page = await api.listOutbox({ cursor, limit: 500 }); + items.push(...page.items); + cursor = page.next_cursor ?? undefined; + } while (cursor); + return items; } export async function loadOverview(): Promise { - const [agents, checks, incidents] = await Promise.all([ + const [agents, checks, openIncidents] = await Promise.all([ loadAgents(), loadChecks(), - api.listIncidents({ state: "open", limit: 500 }), + countOpenIncidents(), ]); - const checksWithLiveness = withLivenessChecks(agents, checks); return { agents: tally(agents), - checks: tally(checksWithLiveness), - openIncidents: incidents.items.length, + checks: tally(checks), + openIncidents, totalAgents: agents.length, - totalChecks: checksWithLiveness.length, + totalChecks: checks.length, }; } +async function countOpenIncidents(): Promise { + let total = 0; + let cursor: string | undefined; + do { + const page = await api.listIncidents({ state: "open", cursor, limit: 500 }); + total += page.items.length; + cursor = page.next_cursor ?? undefined; + } while (cursor); + return total; +} + function tally(xs: { status?: string }[]): Record { return xs.reduce>((acc, x) => { const k = x.status ?? "unknown"; diff --git a/web/src/lib/pagination.ts b/web/src/lib/pagination.ts new file mode 100644 index 0000000..2e8d38b --- /dev/null +++ b/web/src/lib/pagination.ts @@ -0,0 +1,23 @@ +export const PAGE_SIZE_OPTIONS = [10, 50, 100] as const; +export type PageSize = (typeof PAGE_SIZE_OPTIONS)[number]; +export const DEFAULT_PAGE_SIZE: PageSize = 10; + +export function pageCountOf(total: number, pageSize: number): number { + return Math.max(1, Math.ceil(total / pageSize)); +} + +export function clampPage(page: number, pageCount: number): number { + if (page < 1) return 1; + if (page > pageCount) return pageCount; + return page; +} + +export function paginate(items: T[], page: number, pageSize: number): T[] { + const start = (page - 1) * pageSize; + return items.slice(start, start + pageSize); +} + +export function normalizePageSize(value: unknown, fallback: PageSize = DEFAULT_PAGE_SIZE): PageSize { + const n = Number(value); + return (PAGE_SIZE_OPTIONS as readonly number[]).includes(n) ? (n as PageSize) : fallback; +} diff --git a/web/tests/mock-server.mjs b/web/tests/mock-server.mjs index 901103f..074a2aa 100644 --- a/web/tests/mock-server.mjs +++ b/web/tests/mock-server.mjs @@ -55,6 +55,24 @@ const fixtures = { incident_key: "agent-1:load", summary: "load high", }, + { + agent_id: "agent-1", + check_id: "agent_liveness", + status: "ok", + exit_code: 0, + last_observed_at: now, + incident_key: "agent:agent-1:liveness", + summary: "agent is alive", + }, + { + agent_id: "agent-2", + check_id: "agent_liveness", + status: "critical", + exit_code: 2, + last_observed_at: now, + incident_key: "agent:agent-2:liveness", + summary: "agent is dead", + }, ], next_cursor: null, }, @@ -122,6 +140,19 @@ const fixtures = { output_truncated: false, notifications_enabled: true, }, + { + event_id: "00000000-0000-7000-8000-000000000003", + agent_id: "agent-2", + check_id: "agent_liveness", + observed_at: now, + received_at: now, + status: "critical", + exit_code: 2, + duration_ms: 0, + output: "agent is dead", + output_truncated: false, + notifications_enabled: true, + }, ], next_cursor: null, }, diff --git a/web/tests/smoke.spec.ts b/web/tests/smoke.spec.ts index 63aa816..6105da1 100644 --- a/web/tests/smoke.spec.ts +++ b/web/tests/smoke.spec.ts @@ -17,7 +17,7 @@ test("agents list links to detail", async ({ page }) => { await expect(page.locator("thead")).not.toContainText("asc"); await expect(page.locator("thead")).not.toContainText("desc"); await expect(page.locator("thead")).toContainText("↓"); - await expect(page.locator("body")).toContainText("checks 2"); + await expect(page.locator("body")).toContainText("checks 3"); await expect(page.locator("body")).toContainText("warn 1"); await expect(page.getByRole("button", { name: /Timezone: default/ })).toBeVisible(); await page.getByRole("button", { name: /Timezone: default/ }).click(); @@ -73,7 +73,7 @@ test("checks page lists rows", async ({ page }) => { await expect(page.locator("body")).toContainText("host-1"); await expect(page.locator("body")).toContainText("disk ok"); await page.getByRole("link", { name: "Events" }).last().click(); - await expect(page).toHaveURL(/\/checks\/disk\?tab=events$/); + await expect(page).toHaveURL(/\/checks\/disk\?tab=events(?:&events_limit=\d+)?$/); await expect(page.locator("body")).toContainText("12"); await expect(page.locator("thead").last()).not.toContainText("exit"); }); @@ -83,7 +83,6 @@ test("incidents filter renders", async ({ page }) => { await expect(page.locator("body")).toContainText("load high"); await expect(page.locator("thead")).toContainText("HOST"); await expect(page.locator("thead")).not.toContainText("agent_id"); - await expect(page.locator("thead")).toContainText("↓"); await expect(page.locator("tbody")).toContainText("host-1"); await expect(page.locator("tbody")).toContainText("agent_liveness"); await expect(page.locator("tbody")).toContainText("agent is dead"); @@ -91,10 +90,7 @@ test("incidents filter renders", async ({ page }) => { await expect(page).toHaveURL(/\/checks\/agent_liveness$/); await expect(page.locator("body")).toContainText("host-2"); await page.goto("/incidents"); - await page.getByRole("button", { name: /^HOST/ }).click(); - await expect(page).toHaveURL(/\/incidents$/); - await expect(page.locator("thead")).toContainText("↑"); - await page.getByRole("link", { name: "host-1" }).click(); + await page.getByRole("link", { name: "host-1" }).first().click(); await expect(page).toHaveURL(/\/agents\/agent-1$/); });