From 312ea95bcf7975a6d032fb55c31eb04cf01f2b6c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 07:29:21 -0400 Subject: [PATCH] refactor: move host, docker, alert and monitor queries into service modules MCP-3 of milestone 461 "MCP server: Steward as an LLM-inspectable, LLM-operable hub". The /mcp tools run outside Quart's request pipeline, so the queries they share with the pages now live in plain functions that take an open session: - steward/hosts/service.py: uptime, status, monitors, agent metrics, agent overview (stale threshold passed in, not read from current_app), CPU sparks. - plugins/docker/service.py: bucket, container_history, host_map, list_containers, container_events, query_logs. - steward/alerts/service.py: active_alerts (also used by core/reports), list_rules, list_maintenance_windows. - steward/monitors/service.py: monitor_rows (was routes._rows_data). Routes are repointed and render the same data. One behavior fix rides along: the container-log search now escapes % and _ so they match literally instead of acting as LIKE wildcards. Co-Authored-By: Claude Opus 5.5 --- plugins/docker/routes.py | 108 ++---------- plugins/docker/service.py | 99 +++++++++++ steward/alerts/routes.py | 22 +-- steward/alerts/service.py | 36 ++++ steward/core/reports.py | 14 +- steward/hosts/routes.py | 247 ++------------------------- steward/hosts/service.py | 238 ++++++++++++++++++++++++++ steward/monitors/routes.py | 49 +----- steward/monitors/service.py | 52 ++++++ tests/plugins/docker/test_service.py | 18 ++ 10 files changed, 486 insertions(+), 397 deletions(-) create mode 100644 plugins/docker/service.py create mode 100644 steward/alerts/service.py create mode 100644 steward/hosts/service.py create mode 100644 steward/monitors/service.py create mode 100644 tests/plugins/docker/test_service.py diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index c2c522f..e99b031 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -13,11 +13,12 @@ from steward.auth.middleware import require_role from steward.models.users import UserRole from steward.models.hosts import Host from steward.core.time_range import parse_range, DEFAULT_RANGE +from . import service as docker_svc from .dedup import dedup_by_container_id from .swarm_view import build_swarm_services from .models import ( - DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerLog, - DockerMetric, DockerSwarmNode, DockerSwarmService, + DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, + DockerSwarmNode, DockerSwarmService, ) docker_bp = Blueprint("docker", __name__, template_folder="templates") @@ -95,38 +96,6 @@ def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str: ) -def _bucket(values: list[float], target: int = 40) -> list[float]: - """Bucket-average a series down to ~target points (DB-agnostic, in Python). - - Replaces the old SQL strftime() bucketing, which was SQLite-only and broke - on Postgres. Agent cadence is dense, so a multi-hour window is hundreds of - rows — averaging into a readable point count keeps the sparkline's shape. - """ - n = len(values) - if n <= target: - return values - size = (n + target - 1) // target - out: list[float] = [] - for i in range(0, n, size): - chunk = values[i:i + size] - out.append(sum(chunk) / len(chunk)) - return out - - -async def _container_history(db, host_id: str, name: str, since) -> tuple[list, list]: - """Return (cpu_series, mem_series) sparkline-ready for one host's container.""" - rows = (await db.execute( - select(DockerMetric.cpu_pct, DockerMetric.mem_pct) - .where(DockerMetric.host_id == host_id) - .where(DockerMetric.container_name == name) - .where(DockerMetric.scraped_at >= since) - .order_by(DockerMetric.scraped_at) - )).all() - cpu = _bucket([r.cpu_pct or 0.0 for r in rows]) - mem = _bucket([r.mem_pct or 0.0 for r in rows]) - return cpu, mem - - @docker_bp.get("/") @require_role(UserRole.viewer) async def index(): @@ -148,15 +117,6 @@ async def index(): ) -async def _host_map(db, host_ids: set[str]) -> dict[str, Host]: - if not host_ids: - return {} - return { - h.id: h for h in (await db.execute( - select(Host).where(Host.id.in_(host_ids)))).scalars().all() - } - - @docker_bp.get("/rows") @require_role(UserRole.viewer) async def rows(): @@ -165,15 +125,10 @@ async def rows(): since, range_key = parse_range(request.args.get("range")) async with current_app.db_sessionmaker() as db: - containers = list((await db.execute( - select(DockerContainer).order_by( - (DockerContainer.status == "running").desc(), - DockerContainer.name, - ) - )).scalars()) + containers = await docker_svc.list_containers(db) services = list((await db.execute(select(DockerSwarmService))).scalars()) nodes = list((await db.execute(select(DockerSwarmNode))).scalars()) - hosts = await _host_map( + hosts = await docker_svc.host_map( db, {c.host_id for c in containers} | {n.host_id for n in nodes} | {s.host_id for s in services}, @@ -198,7 +153,7 @@ async def rows(): "host_name": host.name if host else c.host_id, "containers": [], "running": 0, "stopped": 0, } - cpu_hist, mem_hist = await _container_history(db, c.host_id, c.name, since) + cpu_hist, mem_hist = await docker_svc.container_history(db, c.host_id, c.name, since) g["containers"].append({ "container": c, "ports": json.loads(c.ports_json) if c.ports_json else [], @@ -267,7 +222,7 @@ async def widget(): select(func.count(func.distinct(DockerEvent.container_name))) .where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since_24h) )).scalar() or 0 - hosts = await _host_map( + hosts = await docker_svc.host_map( db, {c.host_id for c in all_containers} | {n.host_id for n in nodes} | {s.host_id for s in services}, @@ -322,7 +277,7 @@ async def widget_resources(): .where(DockerContainer.status == "running") .order_by(DockerContainer.cpu_pct.desc().nullslast()) )).scalars()))[:limit] - hosts = await _host_map(db, {c.host_id for c in containers}) + hosts = await docker_svc.host_map(db, {c.host_id for c in containers}) rows_data = [ {"c": c, "host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id} @@ -345,14 +300,7 @@ async def host_panel(host_id: str): don't carry an empty card. """ async with current_app.db_sessionmaker() as db: - containers = list((await db.execute( - select(DockerContainer) - .where(DockerContainer.host_id == host_id) - .order_by( - (DockerContainer.status == "running").desc(), - DockerContainer.name, - ) - )).scalars()) + containers = await docker_svc.list_containers(db, host_id) if not containers: return "" @@ -382,15 +330,7 @@ async def container_detail(host_id: str, name: str): async with current_app.db_sessionmaker() as db: c = await db.get(DockerContainer, (host_id, name)) host = await db.get(Host, host_id) - events = [] - if c is not None: - events = list((await db.execute( - select(DockerEvent) - .where(DockerEvent.host_id == host_id) - .where(DockerEvent.container_name == name) - .order_by(DockerEvent.at.desc()) - .limit(50) - )).scalars()) + events = await docker_svc.container_events(db, host_id, name) if c is not None else [] if c is None: return await render_template( @@ -438,7 +378,7 @@ async def swarm(): select(DockerSwarmNode).order_by( DockerSwarmNode.role, DockerSwarmNode.hostname) )).scalars()) - hosts = await _host_map(db, {s.host_id for s in services} | {n.host_id for n in nodes}) + hosts = await docker_svc.host_map(db, {s.host_id for s in services} | {n.host_id for n in nodes}) # ── Group reporting managers into swarms by shared node membership ────────── manager_ids = {s.host_id for s in services} | {n.host_id for n in nodes} @@ -540,7 +480,7 @@ async def disk(): stopped_by_host: dict[str, int] = {} for hid in stopped_rows: stopped_by_host[hid] = stopped_by_host.get(hid, 0) + 1 - hosts = await _host_map(db, {s.host_id for s in summaries}) + hosts = await docker_svc.host_map(db, {s.host_id for s in summaries}) # Which hosts have a linked Ansible target — gates the prune buttons # (a target is required to route the playbook run to that host). host_ids = {s.host_id for s in summaries} @@ -639,7 +579,7 @@ async def container_history(host_id: str, name: str): """HTMX fragment: CPU + memory sparklines for the selected time range.""" since, range_key = parse_range(request.args.get("range")) async with current_app.db_sessionmaker() as db: - cpu_hist, mem_hist = await _container_history(db, host_id, name, since) + cpu_hist, mem_hist = await docker_svc.container_history(db, host_id, name, since) return await render_template( "docker/_container_history.html", sparkline_cpu=_sparkline(cpu_hist, width=320, height=48), @@ -654,26 +594,6 @@ async def container_history(host_id: str, name: str): _LOG_TAIL_DEFAULT = 500 -async def _query_logs(db, host_id: str, name: str, stream: str, query: str, - limit: int) -> list: - """The recent retained lines for one container, newest-first, optionally - filtered by stream and a case-insensitive substring. - - Newest-first is deliberate: the viewer replaces the list on each poll, which - resets scroll to the top — so the latest lines stay visible without any - scroll handling, and older lines are a scroll away. - """ - stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line) - .where(DockerLog.host_id == host_id) - .where(DockerLog.container_name == name)) - if stream in ("stdout", "stderr"): - stmt = stmt.where(DockerLog.stream == stream) - if query: - stmt = stmt.where(DockerLog.line.ilike(f"%{query}%")) - stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit) - return (await db.execute(stmt)).all() - - @docker_bp.get("/container///logs") @require_role(UserRole.viewer) async def container_logs(host_id: str, name: str): @@ -693,7 +613,7 @@ async def container_logs_lines(host_id: str, name: str): stream = request.args.get("stream", "all") query = (request.args.get("q") or "").strip() async with current_app.db_sessionmaker() as db: - rows = await _query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT) + rows = await docker_svc.query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT) return await render_template( "docker/_container_logs_lines.html", lines=[{"ts": r.ts, "stream": r.stream, "line": r.line} for r in rows], diff --git a/plugins/docker/service.py b/plugins/docker/service.py new file mode 100644 index 0000000..c8936ca --- /dev/null +++ b/plugins/docker/service.py @@ -0,0 +1,99 @@ +"""Docker queries shared by the plugin's pages and its MCP tools. + +Each function takes an open session and returns rows or plain data, with no +Quart request context, so the routes and the /mcp tools (which run outside +Quart's pipeline) call the same code. +""" +from __future__ import annotations + +from sqlalchemy import select + +from steward.models.hosts import Host + +from .models import DockerContainer, DockerEvent, DockerLog, DockerMetric + + +def bucket(values: list[float], target: int = 40) -> list[float]: + """Bucket-average a series down to ~target points (DB-agnostic, in Python). + + Replaces the old SQL strftime() bucketing, which was SQLite-only and broke + on Postgres. Agent cadence is dense, so a multi-hour window is hundreds of + rows — averaging into a readable point count keeps the sparkline's shape. + """ + n = len(values) + if n <= target: + return values + size = (n + target - 1) // target + out: list[float] = [] + for i in range(0, n, size): + chunk = values[i:i + size] + out.append(sum(chunk) / len(chunk)) + return out + + +async def container_history(db, host_id: str, name: str, since) -> tuple[list, list]: + """Return (cpu_series, mem_series) sparkline-ready for one host's container.""" + rows = (await db.execute( + select(DockerMetric.cpu_pct, DockerMetric.mem_pct) + .where(DockerMetric.host_id == host_id) + .where(DockerMetric.container_name == name) + .where(DockerMetric.scraped_at >= since) + .order_by(DockerMetric.scraped_at) + )).all() + cpu = bucket([r.cpu_pct or 0.0 for r in rows]) + mem = bucket([r.mem_pct or 0.0 for r in rows]) + return cpu, mem + + +async def host_map(db, host_ids: set[str]) -> dict[str, Host]: + if not host_ids: + return {} + return { + h.id: h for h in (await db.execute( + select(Host).where(Host.id.in_(host_ids)))).scalars().all() + } + + +def escape_like(text: str) -> str: + """Make `text` match literally inside a LIKE/ILIKE pattern (escape char \\).""" + return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +async def list_containers(db, host_id: str | None = None) -> list[DockerContainer]: + """Containers, running first then by name; one host's when host_id is given.""" + stmt = select(DockerContainer) + if host_id is not None: + stmt = stmt.where(DockerContainer.host_id == host_id) + stmt = stmt.order_by((DockerContainer.status == "running").desc(), DockerContainer.name) + return list((await db.execute(stmt)).scalars()) + + +async def container_events(db, host_id: str, name: str, limit: int = 50) -> list[DockerEvent]: + """A container's lifecycle events (start/stop/die/oom/health), newest first.""" + return list((await db.execute( + select(DockerEvent) + .where(DockerEvent.host_id == host_id) + .where(DockerEvent.container_name == name) + .order_by(DockerEvent.at.desc()) + .limit(limit) + )).scalars()) + + +async def query_logs(db, host_id: str, name: str, stream: str, query: str, + limit: int) -> list: + """The recent retained lines for one container, newest-first, optionally + filtered by stream and a case-insensitive substring. + + Newest-first is deliberate: the viewer replaces the list on each poll, which + resets scroll to the top — so the latest lines stay visible without any + scroll handling, and older lines are a scroll away. + """ + stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line) + .where(DockerLog.host_id == host_id) + .where(DockerLog.container_name == name)) + if stream in ("stdout", "stderr"): + stmt = stmt.where(DockerLog.stream == stream) + if query: + stmt = stmt.where(DockerLog.line.ilike(f"%{escape_like(query)}%", escape="\\")) + stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit) + return (await db.execute(stmt)).all() diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index 621c58a..ffbafa8 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -4,6 +4,7 @@ from quart import Blueprint, render_template, request, redirect, url_for, curren from steward.core.audit import log_audit from sqlalchemy import select from steward.auth.middleware import require_role +from steward.alerts import service as alert_svc from steward.ansible import sources as src_module from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, AlertOperator, MaintenanceWindow from steward.models.metrics import PluginMetric @@ -123,20 +124,8 @@ def _parse_ansible_action(form) -> tuple[dict | None, str | None]: @require_role(UserRole.viewer) async def list_alerts(): async with current_app.db_sessionmaker() as db: - result = await db.execute( - select(AlertState, AlertRule) - .join(AlertRule, AlertState.rule_id == AlertRule.id) - .where(AlertState.state.in_([ - AlertStateEnum.firing, AlertStateEnum.acknowledged, AlertStateEnum.pending - ])) - .order_by(AlertState.last_transition_at.desc()) - ) - active = result.all() - - rules_result = await db.execute( - select(AlertRule).order_by(AlertRule.name) - ) - rules = rules_result.scalars().all() + active = await alert_svc.active_alerts(db) + rules = await alert_svc.list_rules(db) return await render_template("alerts/list.html", active=active, rules=rules, operators=_OPERATORS) @@ -319,10 +308,7 @@ async def delete_rule(rule_id: str): async def list_maintenance(): now = datetime.now(timezone.utc) async with current_app.db_sessionmaker() as db: - result = await db.execute( - select(MaintenanceWindow).order_by(MaintenanceWindow.start_at.desc()) - ) - windows = result.scalars().all() + windows = await alert_svc.list_maintenance_windows(db) return await render_template( "alerts/maintenance.html", windows=windows, diff --git a/steward/alerts/service.py b/steward/alerts/service.py new file mode 100644 index 0000000..2dbe4db --- /dev/null +++ b/steward/alerts/service.py @@ -0,0 +1,36 @@ +"""Alert queries shared by the alerts pages and the MCP tools. + +Each function takes an open session and returns rows, with no Quart request +context, so the routes and the /mcp tools (which run outside Quart's pipeline) +call the same code. +""" +from __future__ import annotations + +from sqlalchemy import select + +from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, MaintenanceWindow + +# States that need someone's attention: pending (breaching, not yet for long +# enough), firing, and acknowledged-but-still-breaching. +ACTIVE_STATES = (AlertStateEnum.firing, AlertStateEnum.acknowledged, AlertStateEnum.pending) + + +async def active_alerts(db) -> list: + """(AlertState, AlertRule) rows not yet resolved, latest transition first.""" + return list((await db.execute( + select(AlertState, AlertRule) + .join(AlertRule, AlertState.rule_id == AlertRule.id) + .where(AlertState.state.in_(ACTIVE_STATES)) + .order_by(AlertState.last_transition_at.desc()) + )).all()) + + +async def list_rules(db) -> list[AlertRule]: + return list((await db.execute(select(AlertRule).order_by(AlertRule.name))).scalars()) + + +async def list_maintenance_windows(db) -> list[MaintenanceWindow]: + """Every window, latest start first.""" + return list((await db.execute( + select(MaintenanceWindow).order_by(MaintenanceWindow.start_at.desc()) + )).scalars()) diff --git a/steward/core/reports.py b/steward/core/reports.py index a862371..e65e793 100644 --- a/steward/core/reports.py +++ b/steward/core/reports.py @@ -11,7 +11,8 @@ from email.message import EmailMessage from sqlalchemy import and_, case, func, select -from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, AlertEvent +from steward.alerts import service as alert_svc +from steward.models.alerts import AlertEvent from steward.models.metrics import PluginMetric from steward.models.monitors import Monitor, MonitorResult @@ -51,14 +52,7 @@ async def build_report_data(app) -> dict: ] # ── Active alerts ───────────────────────────────────────────────────── - active_result = await db.execute( - select(AlertState, AlertRule) - .join(AlertRule, AlertState.rule_id == AlertRule.id) - .where(AlertState.state.in_([ - AlertStateEnum.firing, AlertStateEnum.acknowledged, AlertStateEnum.pending, - ])) - .order_by(AlertState.last_transition_at.desc()) - ) + active_rows = await alert_svc.active_alerts(db) active_alerts = [ { "state": state.state.value.upper(), @@ -68,7 +62,7 @@ async def build_report_data(app) -> dict: "operator": rule.operator.value, "threshold": rule.threshold, } - for state, rule in active_result.all() + for state, rule in active_rows ] # ── Alert event counts (7d) ─────────────────────────────────────────── diff --git a/steward/hosts/routes.py b/steward/hosts/routes.py index 1fd430b..4f505a2 100644 --- a/steward/hosts/routes.py +++ b/steward/hosts/routes.py @@ -1,17 +1,15 @@ from __future__ import annotations import asyncio import uuid -from datetime import datetime, timedelta, timezone from quart import Blueprint, render_template, request, redirect, url_for, current_app, session -from sqlalchemy import and_, case, select, func +from sqlalchemy import select from steward.ansible import executor, sources as ansible_src from steward.auth.middleware import require_role from steward.core.audit import log_audit from steward.models.ansible import AnsibleRun, AnsibleRunStatus from steward.models.hosts import Host -from steward.models.monitors import Monitor, MonitorResult from steward.models.users import UserRole -from steward.core.status import _last_n_results, _uptime_by_monitor, _heartbeat_title +from steward.hosts import service as host_svc hosts_bp = Blueprint("hosts", __name__, url_prefix="/hosts") @@ -23,235 +21,18 @@ def _ansible_source_names() -> list[str]: return [] -async def _compute_uptime(db) -> dict[str, dict]: - """Per-host uptime % for 24h/7d/30d, aggregated over the host's monitors. - - Returns {host_id: {"24h": float|None, "7d": float|None, "30d": float|None}}. - """ - now = datetime.now(timezone.utc) - cutoff_30d = now - timedelta(days=30) - cutoff_7d = now - timedelta(days=7) - cutoff_24h = now - timedelta(hours=24) - - result = await db.execute( - select( - Monitor.host_id, - func.count(MonitorResult.id).label("total_30d"), - func.sum(case((MonitorResult.is_up.is_(True), 1), else_=0)).label("up_30d"), - func.sum(case((MonitorResult.checked_at >= cutoff_7d, 1), else_=0)).label("total_7d"), - func.sum(case( - (and_(MonitorResult.checked_at >= cutoff_7d, MonitorResult.is_up.is_(True)), 1), - else_=0, - )).label("up_7d"), - func.sum(case((MonitorResult.checked_at >= cutoff_24h, 1), else_=0)).label("total_24h"), - func.sum(case( - (and_(MonitorResult.checked_at >= cutoff_24h, MonitorResult.is_up.is_(True)), 1), - else_=0, - )).label("up_24h"), - ) - .join(Monitor, Monitor.id == MonitorResult.monitor_id) - .where(Monitor.host_id.isnot(None), MonitorResult.checked_at >= cutoff_30d) - .group_by(Monitor.host_id) - ) - - def _pct(up, total): - return round(float(up) / float(total) * 100, 2) if total else None - - stats: dict[str, dict] = {} - for row in result: - stats[row.host_id] = { - "24h": _pct(row.up_24h, row.total_24h), - "7d": _pct(row.up_7d, row.total_7d), - "30d": _pct(row.up_30d, row.total_30d), - } - return stats - - -async def _host_status(db, host_ids: list[str]) -> dict[str, dict]: - """Per-host status rollup from the latest result of each linked monitor. - - {host_id: {"state": "up"|"down"|"pending", "latency_ms": float|None, - "count": int}}. State is down if ANY linked monitor's latest - result is down (worst-case), up if at least one is up, else pending. - """ - if not host_ids: - return {} - rn = func.row_number().over( - partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc() - ).label("rn") - subq = ( - select(MonitorResult.id, rn) - .join(Monitor, Monitor.id == MonitorResult.monitor_id) - .where(Monitor.host_id.in_(host_ids), Monitor.enabled.is_(True)) - .subquery() - ) - rows = (await db.execute( - select(MonitorResult, Monitor.host_id) - .join(subq, MonitorResult.id == subq.c.id) - .join(Monitor, Monitor.id == MonitorResult.monitor_id) - .where(subq.c.rn == 1) - )).all() - out: dict[str, dict] = {} - for res, host_id in rows: - d = out.setdefault(host_id, {"state": "pending", "latency_ms": None, "count": 0}) - d["count"] += 1 - if not res.is_up: - d["state"] = "down" - elif d["state"] != "down": - d["state"] = "up" - if res.is_up and res.response_ms is not None: - d["latency_ms"] = (res.response_ms if d["latency_ms"] is None - else min(d["latency_ms"], res.response_ms)) - return out - - -async def _host_monitors(db, host_id: str) -> list[dict]: - """Display rows for the monitors linked to one host (host hub section).""" - monitors = (await db.execute( - select(Monitor).where(Monitor.host_id == host_id).order_by(Monitor.name) - )).scalars().all() - ids = [m.id for m in monitors] - recent = await _last_n_results(db, ids) - uptime = await _uptime_by_monitor(db, ids) - data = [] - for m in monitors: - mrows = recent.get(m.id, []) - latest = mrows[-1] if mrows else None - data.append({ - "monitor": m, "latest": latest, - "heartbeat": [{"state": "up" if r.is_up else "down", - "title": _heartbeat_title(m.type, r)} for r in mrows], - "uptime_pct": uptime.get(m.id, {}).get("24h"), - }) - return data - - -async def _agent_metrics_by_host(db, host_names: list[str]) -> dict[str, dict]: - """Latest agent cpu/mem per host name from the generic PluginMetric table. - - Read directly (PluginMetric is a core model) so the Hosts list can show an - at-a-glance agent column without importing the host_agent plugin. - """ - from steward.models.metrics import PluginMetric - if not host_names: - return {} - wanted = ("cpu_pct", "mem_used_pct") - subq = ( - select( - PluginMetric.resource_name, PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("m"), - ) - .where( - PluginMetric.source_module == "host_agent", - PluginMetric.metric_name.in_(wanted), - PluginMetric.resource_name.in_(host_names), - ) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - rows = (await db.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) - & (PluginMetric.metric_name == subq.c.metric_name) - & (PluginMetric.recorded_at == subq.c.m), - ) - )).scalars().all() - out: dict[str, dict] = {} - for r in rows: - out.setdefault(r.resource_name, {})[r.metric_name] = r.value - return out - - -async def _agent_overview_by_host(db, host_names: list[str]) -> dict[str, dict]: - """Per-host agent glance (cpu/mem/root-disk + freshness) from PluginMetric. - - Core-safe (no host_agent import). Freshness compares the latest host-level - sample against the plugin's stale window (config), defaulting to 180s. - """ - from steward.models.metrics import PluginMetric - if not host_names: - return {} - now = datetime.now(timezone.utc) - stale_after = int( - (current_app.config.get("PLUGINS", {}).get("host_agent", {}) or {}) - .get("stale_after_seconds", 180)) - resources = list(host_names) + [h + ":/" for h in host_names] - subq = ( - select( - PluginMetric.resource_name, PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("m"), - ) - .where( - PluginMetric.source_module == "host_agent", - PluginMetric.resource_name.in_(resources), - PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "disk_used_pct")), - ) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - rows = (await db.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) - & (PluginMetric.metric_name == subq.c.metric_name) - & (PluginMetric.recorded_at == subq.c.m), - ).where(PluginMetric.source_module == "host_agent") - )).scalars().all() - out: dict[str, dict] = {} - for r in rows: - if r.resource_name.endswith(":/") and r.metric_name == "disk_used_pct": - out.setdefault(r.resource_name[:-2], {})["disk_root"] = r.value - elif r.resource_name in host_names: - d = out.setdefault(r.resource_name, {}) - d[r.metric_name] = r.value - if d.get("_ts") is None or r.recorded_at > d["_ts"]: - d["_ts"] = r.recorded_at - for d in out.values(): - ts = d.pop("_ts", None) - d["fresh"] = bool(ts and (now - ts).total_seconds() <= stale_after) - return out - - -async def _agent_cpu_sparks_by_host(db, host_names: list[str], hours: int = 1) -> dict[str, str]: - """{host_name: inline-SVG cpu sparkline} over the last `hours` (core-safe). - - A tiny at-a-glance trend in each row — the host-view sparkline, on the widget. - """ - from steward.models.metrics import PluginMetric - from steward.core.status import sparkline_svg - if not host_names: - return {} - cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) - rows = (await db.execute( - select(PluginMetric.resource_name, PluginMetric.value, PluginMetric.recorded_at) - .where( - PluginMetric.source_module == "host_agent", - PluginMetric.metric_name == "cpu_pct", - PluginMetric.resource_name.in_(host_names), - PluginMetric.recorded_at >= cutoff, - ) - .order_by(PluginMetric.recorded_at) - )).all() - by_host: dict[str, list[float]] = {} - for r in rows: - by_host.setdefault(r.resource_name, []).append(r.value) - # Amber line to read as "CPU"; only draw when there are a couple of points. - return { - name: sparkline_svg(vals[-40:], width=64, height=18, stroke="#c8a840") - for name, vals in by_host.items() if len(vals) >= 2 - } - - @hosts_bp.get("/overview/widget") @require_role(UserRole.viewer) async def overview_widget(): """Dashboard widget: unified per-host monitor + agent glance, linking to the hub.""" async with current_app.db_sessionmaker() as db: hosts = (await db.execute(select(Host).order_by(Host.name))).scalars().all() - host_status = await _host_status(db, [h.id for h in hosts]) - uptime = await _compute_uptime(db) + host_status = await host_svc.host_status(db, [h.id for h in hosts]) + uptime = await host_svc.compute_uptime(db) host_names = [h.name for h in hosts] - agent = await _agent_overview_by_host(db, host_names) - cpu_sparks = await _agent_cpu_sparks_by_host(db, host_names) + agent = await host_svc.agent_overview_by_host( + db, host_names, host_svc.agent_stale_after_seconds(current_app.config)) + cpu_sparks = await host_svc.agent_cpu_sparks_by_host(db, host_names) return await render_template( "hosts/overview_widget.html", @@ -266,9 +47,9 @@ async def list_hosts(): async with current_app.db_sessionmaker() as db: result = await db.execute(select(Host).order_by(Host.name)) hosts = result.scalars().all() - host_status = await _host_status(db, [h.id for h in hosts]) - uptime = await _compute_uptime(db) - agent_metrics = await _agent_metrics_by_host(db, [h.name for h in hosts]) + host_status = await host_svc.host_status(db, [h.id for h in hosts]) + uptime = await host_svc.compute_uptime(db) + agent_metrics = await host_svc.agent_metrics_by_host(db, [h.name for h in hosts]) return await render_template( "hosts/list.html", @@ -291,8 +72,8 @@ async def host_detail(host_id: str): select(Host).where(Host.id == host_id))).scalar_one_or_none() if host is None: return "Not found", 404 - monitors = await _host_monitors(db, host_id) - uptime = (await _compute_uptime(db)).get(host_id) + monitors = await host_svc.host_monitors(db, host_id) + uptime = (await host_svc.compute_uptime(db)).get(host_id) linked_target = (await db.execute( select(AnsibleTarget).where(AnsibleTarget.host_id == host_id) .options(selectinload(AnsibleTarget.groups)))).scalar_one_or_none() @@ -500,7 +281,7 @@ async def uptime_page(): async with current_app.db_sessionmaker() as db: result = await db.execute(select(Host).order_by(Host.name)) hosts = result.scalars().all() - uptime = await _compute_uptime(db) + uptime = await host_svc.compute_uptime(db) return await render_template("hosts/uptime.html", hosts=hosts, uptime=uptime) @@ -511,7 +292,7 @@ async def uptime_widget(): async with current_app.db_sessionmaker() as db: result = await db.execute(select(Host).order_by(Host.name)) hosts = result.scalars().all() - uptime = await _compute_uptime(db) + uptime = await host_svc.compute_uptime(db) return await render_template( "hosts/uptime_widget.html", hosts=hosts, diff --git a/steward/hosts/service.py b/steward/hosts/service.py new file mode 100644 index 0000000..da6f2bb --- /dev/null +++ b/steward/hosts/service.py @@ -0,0 +1,238 @@ +"""Host queries shared by the Hosts pages, dashboard widgets and MCP tools. + +Every function takes an open session and returns plain data; none reads +Quart's request context, so the web routes and the /mcp tools (which run +outside Quart's pipeline) call the same code. +""" +from __future__ import annotations +from datetime import datetime, timedelta, timezone + +from sqlalchemy import and_, case, func, select + +from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_monitor +from steward.models.monitors import Monitor, MonitorResult + + +def agent_stale_after_seconds(config) -> int: + """The host_agent plugin's freshness window from app config (default 180s).""" + return int(((config.get("PLUGINS") or {}).get("host_agent") or {}) + .get("stale_after_seconds", 180)) + + +async def compute_uptime(db) -> dict[str, dict]: + """Per-host uptime % for 24h/7d/30d, aggregated over the host's monitors. + + Returns {host_id: {"24h": float|None, "7d": float|None, "30d": float|None}}. + """ + now = datetime.now(timezone.utc) + cutoff_30d = now - timedelta(days=30) + cutoff_7d = now - timedelta(days=7) + cutoff_24h = now - timedelta(hours=24) + + result = await db.execute( + select( + Monitor.host_id, + func.count(MonitorResult.id).label("total_30d"), + func.sum(case((MonitorResult.is_up.is_(True), 1), else_=0)).label("up_30d"), + func.sum(case((MonitorResult.checked_at >= cutoff_7d, 1), else_=0)).label("total_7d"), + func.sum(case( + (and_(MonitorResult.checked_at >= cutoff_7d, MonitorResult.is_up.is_(True)), 1), + else_=0, + )).label("up_7d"), + func.sum(case((MonitorResult.checked_at >= cutoff_24h, 1), else_=0)).label("total_24h"), + func.sum(case( + (and_(MonitorResult.checked_at >= cutoff_24h, MonitorResult.is_up.is_(True)), 1), + else_=0, + )).label("up_24h"), + ) + .join(Monitor, Monitor.id == MonitorResult.monitor_id) + .where(Monitor.host_id.isnot(None), MonitorResult.checked_at >= cutoff_30d) + .group_by(Monitor.host_id) + ) + + def _pct(up, total): + return round(float(up) / float(total) * 100, 2) if total else None + + stats: dict[str, dict] = {} + for row in result: + stats[row.host_id] = { + "24h": _pct(row.up_24h, row.total_24h), + "7d": _pct(row.up_7d, row.total_7d), + "30d": _pct(row.up_30d, row.total_30d), + } + return stats + + +async def host_status(db, host_ids: list[str]) -> dict[str, dict]: + """Per-host status rollup from the latest result of each linked monitor. + + {host_id: {"state": "up"|"down"|"pending", "latency_ms": float|None, + "count": int}}. State is down if ANY linked monitor's latest + result is down (worst-case), up if at least one is up, else pending. + """ + if not host_ids: + return {} + rn = func.row_number().over( + partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc() + ).label("rn") + subq = ( + select(MonitorResult.id, rn) + .join(Monitor, Monitor.id == MonitorResult.monitor_id) + .where(Monitor.host_id.in_(host_ids), Monitor.enabled.is_(True)) + .subquery() + ) + rows = (await db.execute( + select(MonitorResult, Monitor.host_id) + .join(subq, MonitorResult.id == subq.c.id) + .join(Monitor, Monitor.id == MonitorResult.monitor_id) + .where(subq.c.rn == 1) + )).all() + out: dict[str, dict] = {} + for res, host_id in rows: + d = out.setdefault(host_id, {"state": "pending", "latency_ms": None, "count": 0}) + d["count"] += 1 + if not res.is_up: + d["state"] = "down" + elif d["state"] != "down": + d["state"] = "up" + if res.is_up and res.response_ms is not None: + d["latency_ms"] = (res.response_ms if d["latency_ms"] is None + else min(d["latency_ms"], res.response_ms)) + return out + + +async def host_monitors(db, host_id: str) -> list[dict]: + """Display rows for the monitors linked to one host (host hub section).""" + monitors = (await db.execute( + select(Monitor).where(Monitor.host_id == host_id).order_by(Monitor.name) + )).scalars().all() + ids = [m.id for m in monitors] + recent = await _last_n_results(db, ids) + uptime = await _uptime_by_monitor(db, ids) + data = [] + for m in monitors: + mrows = recent.get(m.id, []) + latest = mrows[-1] if mrows else None + data.append({ + "monitor": m, "latest": latest, + "heartbeat": [{"state": "up" if r.is_up else "down", + "title": _heartbeat_title(m.type, r)} for r in mrows], + "uptime_pct": uptime.get(m.id, {}).get("24h"), + }) + return data + + +async def agent_metrics_by_host(db, host_names: list[str]) -> dict[str, dict]: + """Latest agent cpu/mem per host name from the generic PluginMetric table. + + Read directly (PluginMetric is a core model) so the Hosts list can show an + at-a-glance agent column without importing the host_agent plugin. + """ + from steward.models.metrics import PluginMetric + if not host_names: + return {} + wanted = ("cpu_pct", "mem_used_pct") + subq = ( + select( + PluginMetric.resource_name, PluginMetric.metric_name, + func.max(PluginMetric.recorded_at).label("m"), + ) + .where( + PluginMetric.source_module == "host_agent", + PluginMetric.metric_name.in_(wanted), + PluginMetric.resource_name.in_(host_names), + ) + .group_by(PluginMetric.resource_name, PluginMetric.metric_name) + ).subquery() + rows = (await db.execute( + select(PluginMetric).join( + subq, + (PluginMetric.resource_name == subq.c.resource_name) + & (PluginMetric.metric_name == subq.c.metric_name) + & (PluginMetric.recorded_at == subq.c.m), + ) + )).scalars().all() + out: dict[str, dict] = {} + for r in rows: + out.setdefault(r.resource_name, {})[r.metric_name] = r.value + return out + + +async def agent_overview_by_host( + db, host_names: list[str], stale_after_seconds: int = 180, +) -> dict[str, dict]: + """Per-host agent glance (cpu/mem/root-disk + freshness) from PluginMetric. + + Core-safe (no host_agent import). Freshness compares the latest host-level + sample against the host_agent plugin's stale window (stale_after_seconds; + callers pass agent_stale_after_seconds(app.config)). + """ + from steward.models.metrics import PluginMetric + if not host_names: + return {} + now = datetime.now(timezone.utc) + stale_after = int(stale_after_seconds) + resources = list(host_names) + [h + ":/" for h in host_names] + subq = ( + select( + PluginMetric.resource_name, PluginMetric.metric_name, + func.max(PluginMetric.recorded_at).label("m"), + ) + .where( + PluginMetric.source_module == "host_agent", + PluginMetric.resource_name.in_(resources), + PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "disk_used_pct")), + ) + .group_by(PluginMetric.resource_name, PluginMetric.metric_name) + ).subquery() + rows = (await db.execute( + select(PluginMetric).join( + subq, + (PluginMetric.resource_name == subq.c.resource_name) + & (PluginMetric.metric_name == subq.c.metric_name) + & (PluginMetric.recorded_at == subq.c.m), + ).where(PluginMetric.source_module == "host_agent") + )).scalars().all() + out: dict[str, dict] = {} + for r in rows: + if r.resource_name.endswith(":/") and r.metric_name == "disk_used_pct": + out.setdefault(r.resource_name[:-2], {})["disk_root"] = r.value + elif r.resource_name in host_names: + d = out.setdefault(r.resource_name, {}) + d[r.metric_name] = r.value + if d.get("_ts") is None or r.recorded_at > d["_ts"]: + d["_ts"] = r.recorded_at + for d in out.values(): + ts = d.pop("_ts", None) + d["fresh"] = bool(ts and (now - ts).total_seconds() <= stale_after) + return out + + +async def agent_cpu_sparks_by_host(db, host_names: list[str], hours: int = 1) -> dict[str, str]: + """{host_name: inline-SVG cpu sparkline} over the last `hours` (core-safe). + + A tiny at-a-glance trend in each row — the host-view sparkline, on the widget. + """ + from steward.models.metrics import PluginMetric + from steward.core.status import sparkline_svg + if not host_names: + return {} + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) + rows = (await db.execute( + select(PluginMetric.resource_name, PluginMetric.value, PluginMetric.recorded_at) + .where( + PluginMetric.source_module == "host_agent", + PluginMetric.metric_name == "cpu_pct", + PluginMetric.resource_name.in_(host_names), + PluginMetric.recorded_at >= cutoff, + ) + .order_by(PluginMetric.recorded_at) + )).all() + by_host: dict[str, list[float]] = {} + for r in rows: + by_host.setdefault(r.resource_name, []).append(r.value) + # Amber line to read as "CPU"; only draw when there are a couple of points. + return { + name: sparkline_svg(vals[-40:], width=64, height=18, stroke="#c8a840") + for name, vals in by_host.items() if len(vals) >= 2 + } diff --git a/steward/monitors/routes.py b/steward/monitors/routes.py index 36ea7ad..3350173 100644 --- a/steward/monitors/routes.py +++ b/steward/monitors/routes.py @@ -10,13 +10,12 @@ import uuid from datetime import datetime, timezone from quart import Blueprint, current_app, render_template, request, redirect, url_for -from sqlalchemy import select from steward.auth.middleware import require_role from steward.models.users import UserRole from steward.models.monitors import Monitor, MonitorType, MONITOR_TYPE_VALUES -from steward.core.status import _last_n_results, _uptime_by_monitor, _heartbeat_title -from steward.core.time_range import parse_range, DEFAULT_RANGE +from steward.core.time_range import DEFAULT_RANGE +from steward.monitors import service as monitor_svc monitors_bp = Blueprint("monitors", __name__, url_prefix="/monitors") @@ -52,41 +51,6 @@ def _normalise_target(mtype: str, target: str) -> str: return target -async def _rows_data(range_key: str): - """Per-monitor display data for the list / widget.""" - since, range_key = parse_range(range_key) - async with current_app.db_sessionmaker() as db: - monitors = list((await db.execute( - select(Monitor).order_by(Monitor.name) - )).scalars()) - ids = [m.id for m in monitors] - recent = await _last_n_results(db, ids) - uptime = await _uptime_by_monitor(db, ids) - - # Map the selected range to the uptime bucket we surface in the list. - bucket = {"24h": "24h", "7d": "7d", "30d": "30d"}.get(range_key, "24h") - now = datetime.now(timezone.utc) - data = [] - for m in monitors: - rows = recent.get(m.id, []) - latest = rows[-1] if rows else None - tls_days = None - if latest and latest.tls_expires_at: - tls_days = (latest.tls_expires_at - now).total_seconds() / 86400.0 - data.append({ - "monitor": m, - "latest": latest, - "heartbeat": [{"state": "up" if r.is_up else "down", - "title": _heartbeat_title(m.type, r)} for r in rows], - "uptime_pct": uptime.get(m.id, {}).get(bucket), - "tls_days": tls_days, - }) - up = sum(1 for d in data if d["latest"] and d["latest"].is_up) - down = sum(1 for d in data if d["latest"] and not d["latest"].is_up) - pending = sum(1 for d in data if not d["latest"]) - return data, up, down, pending, range_key - - @monitors_bp.get("/") @require_role(UserRole.viewer) async def index(): @@ -102,9 +66,9 @@ async def index(): @monitors_bp.get("/rows") @require_role(UserRole.viewer) async def rows(): - data, up, down, pending, range_key = await _rows_data( - request.args.get("range", DEFAULT_RANGE) - ) + async with current_app.db_sessionmaker() as db: + data, up, down, pending, range_key = await monitor_svc.monitor_rows( + db, request.args.get("range", DEFAULT_RANGE)) return await render_template( "monitors/rows.html", monitor_data=data, up=up, down=down, pending=pending, range_key=range_key, @@ -202,7 +166,8 @@ async def widget(): limit = max(1, min(50, int(request.args.get("limit", 10) or 10))) widget_id = request.args.get("wid", "0") - data, up, down, pending, _ = await _rows_data(DEFAULT_RANGE) + async with current_app.db_sessionmaker() as db: + data, up, down, pending, _ = await monitor_svc.monitor_rows(db, DEFAULT_RANGE) if type_filter in MONITOR_TYPE_VALUES: data = [d for d in data if d["monitor"].type == type_filter] if show_down_only: diff --git a/steward/monitors/service.py b/steward/monitors/service.py new file mode 100644 index 0000000..2c3d92e --- /dev/null +++ b/steward/monitors/service.py @@ -0,0 +1,52 @@ +"""Monitor queries shared by the monitors pages and the MCP tools. + +Each function takes an open session, with no Quart request context, so the +routes and the /mcp tools (which run outside Quart's pipeline) call the same +code. +""" +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import select + +from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_monitor +from steward.core.time_range import parse_range +from steward.models.monitors import Monitor + + +async def monitor_rows(db, range_key: str): + """Per-monitor state: latest result, heartbeat strip, uptime and TLS days. + + Returns (rows, up, down, pending, range_key). `range_key` picks the uptime + window surfaced (24h/7d/30d; anything else falls back to 24h) and comes + back normalised. + """ + _since, range_key = parse_range(range_key) + monitors = list((await db.execute(select(Monitor).order_by(Monitor.name))).scalars()) + ids = [m.id for m in monitors] + recent = await _last_n_results(db, ids) + uptime = await _uptime_by_monitor(db, ids) + + # Map the selected range to the uptime bucket we surface in the list. + bucket = {"24h": "24h", "7d": "7d", "30d": "30d"}.get(range_key, "24h") + now = datetime.now(timezone.utc) + data = [] + for m in monitors: + rows = recent.get(m.id, []) + latest = rows[-1] if rows else None + tls_days = None + if latest and latest.tls_expires_at: + tls_days = (latest.tls_expires_at - now).total_seconds() / 86400.0 + data.append({ + "monitor": m, + "latest": latest, + "heartbeat": [{"state": "up" if r.is_up else "down", + "title": _heartbeat_title(m.type, r)} for r in rows], + "uptime_pct": uptime.get(m.id, {}).get(bucket), + "tls_days": tls_days, + }) + up = sum(1 for d in data if d["latest"] and d["latest"].is_up) + down = sum(1 for d in data if d["latest"] and not d["latest"].is_up) + pending = sum(1 for d in data if not d["latest"]) + return data, up, down, pending, range_key diff --git a/tests/plugins/docker/test_service.py b/tests/plugins/docker/test_service.py new file mode 100644 index 0000000..93c9ef0 --- /dev/null +++ b/tests/plugins/docker/test_service.py @@ -0,0 +1,18 @@ +"""Unit tests for the docker plugin's shared query helpers (no DB).""" +from plugins.docker import service as svc + + +def test_escape_like_makes_wildcards_literal(): + # A log search for "100%" or "user_id" must match those characters, not + # treat them as LIKE wildcards. + assert svc.escape_like("100%") == "100\\%" + assert svc.escape_like("user_id") == "user\\_id" + assert svc.escape_like("C:\\temp") == "C:\\\\temp" + assert svc.escape_like("plain text") == "plain text" + + +def test_bucket_keeps_short_series_and_averages_long_ones(): + assert svc.bucket([1.0, 2.0, 3.0]) == [1.0, 2.0, 3.0] + out = svc.bucket([float(i) for i in range(400)], target=40) + assert len(out) == 40 + assert out[0] == sum(range(10)) / 10