diff --git a/host_agent/__init__.py b/host_agent/__init__.py new file mode 100644 index 0000000..6a7b229 --- /dev/null +++ b/host_agent/__init__.py @@ -0,0 +1,23 @@ +# plugins/host_agent/__init__.py +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from quart import Quart + +_app: "Quart | None" = None + + +def setup(app: "Quart") -> None: + global _app + _app = app + + +def get_scheduled_tasks() -> list: + from .scheduler import make_task + return [make_task(_app)] + + +def get_blueprint(): + from .routes import host_agent_bp + return host_agent_bp diff --git a/host_agent/agent.py b/host_agent/agent.py new file mode 100644 index 0000000..cbb3561 --- /dev/null +++ b/host_agent/agent.py @@ -0,0 +1,370 @@ +# plugins/host_agent/agent.py +"""Roundtable host agent — pushes resource metrics to a Roundtable instance. + +Python 3.8+ stdlib only. Target ~300 lines. Served to targets at +GET /plugins/host_agent/agent.py. +""" +from __future__ import annotations + +import json +import os +import shutil +import signal +import socket +import sys +import time +import urllib.error +import urllib.request +from collections import deque +from datetime import datetime, timezone + +AGENT_VERSION = "1.0.0" + + +class ConfigError(Exception): + pass + + +REQUIRED_KEYS = ("url", "token") +INT_KEYS = ("interval_seconds",) +LIST_KEYS = ("mounts",) + + +def read_config(path: str) -> dict: + """Parse a flat `key = value` config file.""" + cfg: dict = {} + try: + with open(path, "r", encoding="utf-8") as f: + for lineno, raw in enumerate(f, 1): + line = raw.strip() + if not line or line.startswith("#"): + continue + if "=" not in line: + raise ConfigError(f"{path}:{lineno}: expected 'key = value'") + key, _, value = line.partition("=") + key = key.strip() + value = value.strip() + if key in INT_KEYS: + try: + cfg[key] = int(value) + except ValueError: + raise ConfigError(f"{path}:{lineno}: {key} must be int") + elif key in LIST_KEYS: + cfg[key] = [v.strip() for v in value.split(",") if v.strip()] + else: + cfg[key] = value + except FileNotFoundError: + raise ConfigError(f"{path}: not found") + + missing = [k for k in REQUIRED_KEYS if k not in cfg] + if missing: + raise ConfigError(f"{path}: missing required key(s): {', '.join(missing)}") + + cfg.setdefault("interval_seconds", 30) + return cfg + + +# ─── collectors ────────────────────────────────────────────────────────────── + +STAT_PATH = "/proc/stat" +MEMINFO_PATH = "/proc/meminfo" +LOADAVG_PATH = "/proc/loadavg" +UPTIME_PATH = "/proc/uptime" +OS_RELEASE_PATH = "/etc/os-release" + + +def _read_file(path: str) -> str: + with open(path, "r", encoding="utf-8") as f: + return f.read() + + +def _parse_cpu_line(stat_text: str) -> tuple[int, int]: + """Return (total_jiffies, idle_jiffies) for the aggregate cpu line.""" + for line in stat_text.splitlines(): + if line.startswith("cpu "): + parts = line.split() + fields = [int(x) for x in parts[1:]] + idle = fields[3] + (fields[4] if len(fields) > 4 else 0) # idle + iowait + total = sum(fields) + return total, idle + raise RuntimeError("no aggregate cpu line in /proc/stat") + + +def collect_cpu(sample_window: float = 0.2) -> float: + """Return CPU utilization % over sample_window seconds.""" + t1, i1 = _parse_cpu_line(_read_file(STAT_PATH)) + time.sleep(sample_window) + t2, i2 = _parse_cpu_line(_read_file(STAT_PATH)) + dt = t2 - t1 + di = i2 - i1 + if dt <= 0: + return 0.0 + return round(100.0 * (dt - di) / dt, 2) + + +def collect_memory() -> dict: + info: dict[str, int] = {} + for line in _read_file(MEMINFO_PATH).splitlines(): + if ":" not in line: + continue + key, _, rest = line.partition(":") + parts = rest.strip().split() + if not parts: + continue + try: + value_kb = int(parts[0]) + except ValueError: + continue + info[key] = value_kb * 1024 # bytes + total = info.get("MemTotal", 0) + available = info.get("MemAvailable", 0) + swap_total = info.get("SwapTotal", 0) + swap_free = info.get("SwapFree", 0) + return { + "total_bytes": total, + "used_bytes": max(total - available, 0), + "available_bytes": available, + "swap_used_bytes": max(swap_total - swap_free, 0), + } + + +def collect_storage(mounts: list[str]) -> list[dict]: + out: list[dict] = [] + for m in mounts: + try: + u = shutil.disk_usage(m) + except (FileNotFoundError, PermissionError): + continue + out.append({"mount": m, "total_bytes": u.total, "used_bytes": u.used}) + return out + + +def collect_load() -> dict: + parts = _read_file(LOADAVG_PATH).split() + return {"1m": float(parts[0]), "5m": float(parts[1]), "15m": float(parts[2])} + + +def collect_uptime() -> int: + return int(float(_read_file(UPTIME_PATH).split()[0])) + + +def collect_metadata() -> dict: + u = os.uname() + distro = "unknown" + try: + for line in _read_file(OS_RELEASE_PATH).splitlines(): + if line.startswith("PRETTY_NAME="): + distro = line.split("=", 1)[1].strip().strip('"') + break + except (OSError, FileNotFoundError): + pass + return {"kernel": u.release, "distro": distro, "arch": u.machine} + + +def default_mounts() -> list[str]: + """Return real mount points from /proc/mounts, skipping pseudo filesystems.""" + skip_types = {"tmpfs", "devtmpfs", "proc", "sysfs", "cgroup", "cgroup2", + "overlay", "squashfs", "ramfs", "devpts", "mqueue", "pstore", + "securityfs", "debugfs", "tracefs", "hugetlbfs", "fusectl", + "configfs", "bpf", "autofs", "efivarfs", "binfmt_misc"} + mounts: list[str] = [] + try: + with open("/proc/mounts", "r") as f: + for line in f: + parts = line.split() + if len(parts) < 3: + continue + if parts[2] in skip_types: + continue + mounts.append(parts[1]) + except OSError: + return ["/"] + return mounts or ["/"] + + +# ─── ring buffer ───────────────────────────────────────────────────────────── + + +class RingBuffer: + def __init__(self, maxlen: int = 20) -> None: + self._dq: deque = deque(maxlen=maxlen) + + def push(self, item) -> None: + self._dq.append(item) + + def drain(self) -> list: + out = list(self._dq) + self._dq.clear() + return out + + def __len__(self) -> int: + return len(self._dq) + + +# ─── payload ───────────────────────────────────────────────────────────────── + + +def build_sample(mounts: list[str]) -> dict: + """Collect one full sample. Partial samples allowed if a collector fails.""" + sample: dict = {"ts": datetime.now(timezone.utc).isoformat()} + try: + sample["cpu_pct"] = collect_cpu() + except Exception: + sample["cpu_pct"] = None + try: + sample["mem"] = collect_memory() + except Exception: + sample["mem"] = None + try: + sample["load"] = collect_load() + except Exception: + sample["load"] = None + try: + sample["uptime_secs"] = collect_uptime() + except Exception: + sample["uptime_secs"] = None + try: + sample["storage"] = collect_storage(mounts) + except Exception: + sample["storage"] = [] + return sample + + +def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict: + return { + "agent_version": AGENT_VERSION, + "hostname": hostname, + "metadata": metadata, + "samples": samples, + } + + +# ─── POST + backoff ────────────────────────────────────────────────────────── + + +BACKOFF_CAP = 300 + + +def next_backoff(current: int) -> int: + if current <= 0: + return 30 + return min(current * 2, BACKOFF_CAP) + + +def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]: + body = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url.rstrip("/") + "/plugins/host_agent/ingest", + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {token}", + "User-Agent": f"roundtable-host-agent/{AGENT_VERSION}", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + return (200 <= resp.status < 300, resp.status) + except urllib.error.HTTPError as e: + return (False, e.code) + except (urllib.error.URLError, TimeoutError, socket.timeout, OSError): + return (False, None) + + +# ─── main loop ─────────────────────────────────────────────────────────────── + + +_reload_requested = False +_shutdown_requested = False + + +def _handle_hup(_signum, _frame): + global _reload_requested + _reload_requested = True + + +def _handle_term(_signum, _frame): + global _shutdown_requested + _shutdown_requested = True + + +def _log(level: str, msg: str) -> None: + sys.stderr.write(f"[{level}] {msg}\n") + sys.stderr.flush() + + +def main_loop(conf_path: str) -> int: + global _reload_requested, _shutdown_requested + signal.signal(signal.SIGHUP, _handle_hup) + signal.signal(signal.SIGTERM, _handle_term) + + try: + cfg = read_config(conf_path) + except ConfigError as e: + _log("ERROR", str(e)) + return 2 + + metadata = collect_metadata() + hostname = cfg.get("hostname") or socket.gethostname() + mounts = cfg.get("mounts") or default_mounts() + buffer = RingBuffer(maxlen=20) + backoff = 0 + + _log("INFO", f"roundtable-host-agent {AGENT_VERSION} starting " + f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)") + + while not _shutdown_requested: + if _reload_requested: + try: + cfg = read_config(conf_path) + mounts = cfg.get("mounts") or default_mounts() + _log("INFO", "config reloaded") + except ConfigError as e: + _log("ERROR", f"reload failed: {e}") + _reload_requested = False + + sample = build_sample(mounts) + buffered = buffer.drain() + payload = build_payload( + samples=buffered + [sample], + hostname=hostname, + metadata=metadata, + ) + ok, status = post_payload(cfg["url"], cfg["token"], payload) + + if ok: + if buffered: + _log("INFO", f"flushed {len(buffered)} buffered samples") + backoff = 0 + sleep_for = cfg["interval_seconds"] + else: + if status == 400: + _log("ERROR", "server rejected payload (400) — dropping sample") + elif status == 401: + _log("ERROR", "token rejected (401) — check config + UI") + buffer.push(sample) + else: + _log("WARN", f"POST failed (status={status}); buffering") + for s in buffered: + buffer.push(s) + buffer.push(sample) + backoff = next_backoff(backoff) + sleep_for = backoff + + slept = 0.0 + while slept < sleep_for and not _shutdown_requested and not _reload_requested: + time.sleep(min(1.0, sleep_for - slept)) + slept += 1.0 + + _log("INFO", "SIGTERM — flushing and exiting") + final = buffer.drain() + if final: + post_payload(cfg["url"], cfg["token"], + build_payload(final, hostname, metadata)) + return 0 + + +if __name__ == "__main__": + conf = os.environ.get("ROUNDTABLE_AGENT_CONFIG", "/etc/roundtable-agent.conf") + sys.exit(main_loop(conf)) diff --git a/host_agent/migrations/__init__.py b/host_agent/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/host_agent/migrations/env.py b/host_agent/migrations/env.py new file mode 100644 index 0000000..35d343a --- /dev/null +++ b/host_agent/migrations/env.py @@ -0,0 +1,70 @@ +# plugins/host_agent/migrations/env.py +"""Alembic env.py for the host_agent plugin.""" +import asyncio +import os +import sys +from pathlib import Path +from logging.config import fileConfig + +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config +from alembic import context + +sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) + +from roundtable.models.base import Base +import roundtable.models # noqa: F401 +from plugins.host_agent.models import HostAgentRegistration # noqa: F401 + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def _get_url() -> str: + import yaml + cfg_path = os.environ.get("ROUNDTABLE_CONFIG", "config.yaml") + try: + with open(cfg_path) as f: + cfg = yaml.safe_load(f) or {} + url = cfg.get("database", {}).get("url", "") + except FileNotFoundError: + url = "" + return os.environ.get("ROUNDTABLE_DATABASE__URL", url) + + +def run_migrations_offline() -> None: + context.configure( + url=_get_url(), target_metadata=target_metadata, + literal_binds=True, dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + cfg = config.get_section(config.config_ini_section, {}) + cfg["sqlalchemy.url"] = _get_url() + connectable = async_engine_from_config(cfg, prefix="sqlalchemy.", poolclass=pool.NullPool) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/host_agent/migrations/versions/__init__.py b/host_agent/migrations/versions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/host_agent/migrations/versions/host_agent_001_initial.py b/host_agent/migrations/versions/host_agent_001_initial.py new file mode 100644 index 0000000..d241257 --- /dev/null +++ b/host_agent/migrations/versions/host_agent_001_initial.py @@ -0,0 +1,36 @@ +# plugins/host_agent/migrations/versions/host_agent_001_initial.py +"""host_agent plugin initial tables + +Revision ID: host_agent_001_initial +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "host_agent_001_initial" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = "host_agent" +depends_on: Union[str, Sequence[str], None] = ("0004_app_settings",) + + +def upgrade() -> None: + op.create_table( + "host_agent_registrations", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False, unique=True), + sa.Column("token_hash", sa.String(64), nullable=False, unique=True, index=True), + sa.Column("token_created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("agent_version", sa.String(32), nullable=True), + sa.Column("kernel", sa.String(128), nullable=True), + sa.Column("distro", sa.String(128), nullable=True), + sa.Column("arch", sa.String(32), nullable=True), + sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("host_agent_registrations") diff --git a/host_agent/models.py b/host_agent/models.py new file mode 100644 index 0000000..0c92065 --- /dev/null +++ b/host_agent/models.py @@ -0,0 +1,32 @@ +# plugins/host_agent/models.py +from __future__ import annotations +import uuid +from datetime import datetime, timezone +from sqlalchemy import Column, String, DateTime, ForeignKey +from roundtable.models.base import Base + + +def _uuid() -> str: + return str(uuid.uuid4()) + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class HostAgentRegistration(Base): + __tablename__ = "host_agent_registrations" + + id = Column(String(36), primary_key=True, default=_uuid) + host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"), + unique=True, nullable=False) + token_hash = Column(String(64), nullable=False, unique=True, index=True) + token_created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + agent_version = Column(String(32), nullable=True) + kernel = Column(String(128), nullable=True) + distro = Column(String(128), nullable=True) + arch = Column(String(32), nullable=True) + last_seen_at = Column(DateTime(timezone=True), nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + updated_at = Column(DateTime(timezone=True), nullable=False, + default=_utcnow, onupdate=_utcnow) diff --git a/host_agent/plugin.yaml b/host_agent/plugin.yaml new file mode 100644 index 0000000..1ce36c5 --- /dev/null +++ b/host_agent/plugin.yaml @@ -0,0 +1,19 @@ +# plugins/host_agent/plugin.yaml +name: host_agent +version: "1.0.0" +description: "Remote Linux host resource monitoring via a lightweight Python push agent (CPU, memory, storage, load, uptime)" +author: "Roundtable" +license: "MIT" +min_app_version: "0.1.0" +repository_url: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins" +homepage: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins/src/branch/main/host_agent" +tags: + - host + - monitoring + - cpu + - memory + - storage + +config: + stale_after_seconds: 180 + default_interval_seconds: 30 diff --git a/host_agent/routes.py b/host_agent/routes.py new file mode 100644 index 0000000..d897676 --- /dev/null +++ b/host_agent/routes.py @@ -0,0 +1,397 @@ +# plugins/host_agent/routes.py +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone +from typing import Any + +from pathlib import Path + +import secrets +from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for +from roundtable.auth.middleware import require_role +from roundtable.models.users import UserRole +from sqlalchemy import select, func +from datetime import timedelta + +from roundtable.models.hosts import Host +from roundtable.models.metrics import PluginMetric +from .models import HostAgentRegistration + +host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates") + +SOURCE_MODULE = "host_agent" + + +def _hash_token(raw: str) -> str: + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _parse_ts(ts: str) -> datetime: + if ts.endswith("Z"): + ts = ts[:-1] + "+00:00" + return datetime.fromisoformat(ts) + + +def _expand_sample_to_metrics( + sample: dict, host_name: str, recorded_at: datetime +) -> list[PluginMetric]: + rows: list[PluginMetric] = [] + + def row(metric: str, resource: str, value: float) -> None: + rows.append(PluginMetric( + source_module=SOURCE_MODULE, + resource_name=resource, + metric_name=metric, + value=float(value), + recorded_at=recorded_at, + )) + + if sample.get("cpu_pct") is not None: + row("cpu_pct", host_name, sample["cpu_pct"]) + + mem = sample.get("mem") or {} + if mem.get("total_bytes"): + total = mem["total_bytes"] + available = mem.get("available_bytes", 0) + used_pct = 100.0 * (total - available) / total if total else 0.0 + row("mem_used_pct", host_name, used_pct) + row("mem_available_bytes", host_name, available) + row("swap_used_bytes", host_name, mem.get("swap_used_bytes", 0)) + + load = sample.get("load") or {} + for key in ("1m", "5m", "15m"): + if key in load: + row(f"load_{key}", host_name, load[key]) + + if sample.get("uptime_secs") is not None: + row("uptime_secs", host_name, sample["uptime_secs"]) + + worst_pct = 0.0 + for disk in sample.get("storage") or []: + total = disk.get("total_bytes", 0) + used = disk.get("used_bytes", 0) + pct = 100.0 * used / total if total else 0.0 + mount_res = f"{host_name}:{disk['mount']}" + row("disk_used_pct", mount_res, pct) + row("disk_used_bytes", mount_res, used) + row("disk_total_bytes", mount_res, total) + if pct > worst_pct: + worst_pct = pct + if sample.get("storage"): + row("disk_used_pct_worst", host_name, worst_pct) + + return rows + + +def _error(status: int, code: str, detail: str | None = None) -> tuple[Any, int]: + body: dict = {"ok": False, "error": code} + if detail: + body["detail"] = detail + return jsonify(body), status + + +@host_agent_bp.post("/ingest") +async def ingest(): + auth = request.headers.get("Authorization", "") + if not auth.startswith("Bearer "): + return _error(401, "invalid_token") + raw = auth[len("Bearer "):].strip() + if not raw: + return _error(401, "invalid_token") + + try: + payload = await request.get_json(force=True) + except Exception: + return _error(400, "malformed_payload", "invalid JSON") + if not isinstance(payload, dict) or "samples" not in payload: + return _error(400, "malformed_payload", "missing 'samples'") + + samples = payload.get("samples") or [] + if not isinstance(samples, list) or not samples: + return _error(400, "malformed_payload", "'samples' must be a non-empty list") + + token_hash = _hash_token(raw) + async with current_app.db_sessionmaker() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.token_hash == token_hash) + )).scalar_one_or_none() + if reg is None: + return _error(401, "invalid_token") + host = (await session.execute( + select(Host).where(Host.id == reg.host_id) + )).scalar_one_or_none() + if host is None: + return _error(401, "invalid_token") + + accepted = 0 + latest_ts: datetime | None = None + for sample in samples: + try: + recorded_at = _parse_ts(sample["ts"]) + except (KeyError, ValueError): + continue + metrics = _expand_sample_to_metrics(sample, host.name, recorded_at) + for m in metrics: + session.add(m) + accepted += 1 + if latest_ts is None or recorded_at > latest_ts: + latest_ts = recorded_at + + if accepted == 0: + await session.rollback() + return _error(400, "malformed_payload", "no valid samples") + + md = payload.get("metadata") or {} + changed = False + if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at): + reg.last_seen_at = latest_ts + changed = True + version = payload.get("agent_version") + if version and reg.agent_version != version: + reg.agent_version = version + changed = True + for field in ("kernel", "distro", "arch"): + if field in md and getattr(reg, field) != md[field]: + setattr(reg, field, md[field]) + changed = True + if changed: + reg.updated_at = datetime.now(timezone.utc) + + if latest_ts: + skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds()) + if skew > 300: + current_app.logger.warning( + "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name) + + await session.commit() + + return jsonify({"ok": True, "samples_accepted": accepted}), 200 + + +AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py" + + +def _agent_version() -> str: + try: + for line in AGENT_SOURCE_PATH.read_text().splitlines(): + if line.startswith("AGENT_VERSION"): + return line.split("=", 1)[1].strip().strip('"').strip("'") + except OSError: + pass + return "unknown" + + +@host_agent_bp.get("/install.sh") +async def install_script(): + token = request.args.get("token", "") + if not token: + return _error(401, "invalid_token") + async with current_app.db_sessionmaker() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.token_hash == _hash_token(token)) + )).scalar_one_or_none() + if reg is None: + return _error(401, "invalid_token") + host = (await session.execute( + select(Host).where(Host.id == reg.host_id) + )).scalar_one_or_none() + if host is None: + return _error(401, "invalid_token") + + url = request.host_url.rstrip("/") + + rendered = await render_template( + "install.sh.j2", + url=url, + token=token, + agent_version=_agent_version(), + host_name=host.name, + host_address=host.address, + ) + return Response(rendered, mimetype="text/plain") + + +@host_agent_bp.get("/agent.py") +async def agent_source(): + try: + body = AGENT_SOURCE_PATH.read_text(encoding="utf-8") + except OSError: + return _error(500, "agent_missing") + return Response(body, mimetype="text/x-python") + + +@host_agent_bp.get("/widget") +async def widget_table(): + """Fleet-glance table: one row per monitored host, latest metrics.""" + async with current_app.db_sessionmaker() as session: + regs = (await session.execute(select(HostAgentRegistration))).scalars().all() + host_ids = [r.host_id for r in regs] + hosts = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_(host_ids)) + )).scalars().all() + } if host_ids else {} + + subq = ( + select( + PluginMetric.resource_name, + PluginMetric.metric_name, + func.max(PluginMetric.recorded_at).label("max_ts"), + ) + .where(PluginMetric.source_module == SOURCE_MODULE) + .group_by(PluginMetric.resource_name, PluginMetric.metric_name) + ).subquery() + latest_rows = (await session.execute( + select(PluginMetric).join( + subq, + (PluginMetric.resource_name == subq.c.resource_name) & + (PluginMetric.metric_name == subq.c.metric_name) & + (PluginMetric.recorded_at == subq.c.max_ts), + ).where(PluginMetric.source_module == SOURCE_MODULE) + )).scalars().all() + + latest: dict[str, dict[str, float]] = {} + for row in latest_rows: + if ":" in row.resource_name: + continue + latest.setdefault(row.resource_name, {})[row.metric_name] = row.value + + rows = [] + for reg in regs: + host = hosts.get(reg.host_id) + if host is None: + continue + m = latest.get(host.name, {}) + rows.append({ + "host": host, + "reg": reg, + "cpu_pct": m.get("cpu_pct"), + "mem_used_pct": m.get("mem_used_pct"), + "disk_worst": m.get("disk_used_pct_worst"), + "load_1m": m.get("load_1m"), + }) + + return await render_template("widget_table.html", rows=rows) + + +@host_agent_bp.get("/widget/history") +async def widget_history(): + host_id = request.args.get("host_id", "") + hours = int(request.args.get("hours", "6")) + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) + + async with current_app.db_sessionmaker() as session: + host = (await session.execute( + select(Host).where(Host.id == host_id))).scalar_one_or_none() + if host is None: + return _error(404, "not_found") + points = (await session.execute( + select(PluginMetric).where( + PluginMetric.source_module == SOURCE_MODULE, + PluginMetric.resource_name == host.name, + PluginMetric.metric_name.in_( + ("cpu_pct", "mem_used_pct", "disk_used_pct_worst")), + PluginMetric.recorded_at >= cutoff, + ).order_by(PluginMetric.recorded_at) + )).scalars().all() + + series: dict[str, list[dict]] = {"cpu_pct": [], "mem_used_pct": [], "disk_used_pct_worst": []} + for p in points: + series[p.metric_name].append({"t": p.recorded_at.isoformat(), "v": p.value}) + + return await render_template("widget_history.html", host=host, series=series, hours=hours) + + +def _new_token_pair() -> tuple[str, str]: + raw = secrets.token_urlsafe(32) + return raw, _hash_token(raw) + + +@host_agent_bp.get("/settings/") +@require_role(UserRole.admin) +async def settings_list(): + async with current_app.db_sessionmaker() as session: + regs = (await session.execute(select(HostAgentRegistration))).scalars().all() + hosts_by_id = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all() + } if regs else {} + + new_token = request.args.get("new_token") + new_host_id = request.args.get("host_id") + install_url = None + if new_token and new_host_id: + install_url = f"{request.host_url.rstrip('/')}/plugins/host_agent/install.sh?token={new_token}" + + return await render_template( + "settings_list.html", + registrations=[ + {"reg": r, "host": hosts_by_id.get(r.host_id)} for r in regs + ], + new_token=new_token, + install_url=install_url, + ) + + +@host_agent_bp.post("/settings/add-host") +@require_role(UserRole.admin) +async def add_host(): + form = await request.form + name = (form.get("name") or "").strip() + address = (form.get("address") or "").strip() + if not name: + return _error(400, "missing_name") + + async with current_app.db_sessionmaker() as session: + host = (await session.execute( + select(Host).where(Host.name == name))).scalar_one_or_none() + if host is None: + host = Host(name=name, address=address) + session.add(host) + await session.flush() + existing = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host.id))).scalar_one_or_none() + if existing is not None: + await session.rollback() + return _error(400, "already_registered") + + raw, hashed = _new_token_pair() + reg = HostAgentRegistration(host_id=host.id, token_hash=hashed) + session.add(reg) + await session.commit() + host_id = host.id + + return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host_id}") + + +@host_agent_bp.post("/settings//rotate-token") +@require_role(UserRole.admin) +async def rotate_token(host_id: str): + async with current_app.db_sessionmaker() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + if reg is None: + return _error(404, "not_found") + raw, hashed = _new_token_pair() + reg.token_hash = hashed + reg.token_created_at = datetime.now(timezone.utc) + await session.commit() + return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host_id}") + + +@host_agent_bp.post("/settings//delete") +@require_role(UserRole.admin) +async def delete_registration(host_id: str): + async with current_app.db_sessionmaker() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + if reg is not None: + await session.delete(reg) + await session.commit() + return redirect(url_for("host_agent.settings_list")) diff --git a/host_agent/scheduler.py b/host_agent/scheduler.py new file mode 100644 index 0000000..1d750f4 --- /dev/null +++ b/host_agent/scheduler.py @@ -0,0 +1,78 @@ +# plugins/host_agent/scheduler.py +from __future__ import annotations +import logging +from datetime import datetime, timedelta, timezone +from typing import Iterable + +from sqlalchemy import select + +from roundtable.core.scheduler import ScheduledTask +from roundtable.models.hosts import Host +from .models import HostAgentRegistration + +logger = logging.getLogger(__name__) + + +def _filter_stale( + regs: Iterable, + *, + now: datetime, + stale_after_seconds: int, +) -> list: + """Pure staleness filter: returns the subset with last_seen_at strictly + older than (now - stale_after_seconds). Rows with last_seen_at=None are + never stale (they are unregistered-in-practice).""" + cutoff = now - timedelta(seconds=stale_after_seconds) + return [ + r for r in regs + if r.last_seen_at is not None and r.last_seen_at < cutoff + ] + + +async def find_stale_registrations(app, stale_after_seconds: int = 180) -> list[dict]: + async with app.db_sessionmaker() as session: + all_regs = (await session.execute( + select(HostAgentRegistration) + )).scalars().all() + stale_rows = _filter_stale( + all_regs, + now=datetime.now(timezone.utc), + stale_after_seconds=stale_after_seconds, + ) + if not stale_rows: + return [] + hosts = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_([r.host_id for r in stale_rows])) + )).scalars().all() + } + return [ + { + "host_id": r.host_id, + "host_name": hosts[r.host_id].name if r.host_id in hosts else "?", + "last_seen_at": r.last_seen_at, + } + for r in stale_rows + ] + + +def make_task(app) -> ScheduledTask: + async def _check_stale(): + try: + stale = await find_stale_registrations(app) + except Exception: + logger.exception("host_agent stale check failed") + return + if stale: + logger.info( + "host_agent: %d stale agent(s): %s", + len(stale), + [s["host_name"] for s in stale], + ) + + return ScheduledTask( + name="host_agent_stale_check", + coro_factory=_check_stale, + interval_seconds=60, + run_on_startup=False, + ) diff --git a/host_agent/templates/install.sh.j2 b/host_agent/templates/install.sh.j2 new file mode 100644 index 0000000..34789e7 --- /dev/null +++ b/host_agent/templates/install.sh.j2 @@ -0,0 +1,81 @@ +#!/bin/sh +# Roundtable host agent installer +# Generated for: {{ host_name }} ({{ host_address }}) +# Roundtable URL: {{ url }} +set -e + +ROUNDTABLE_URL="{{ url }}" +AGENT_TOKEN="{{ token }}" +AGENT_VERSION="{{ agent_version }}" + +AGENT_USER="roundtable-agent" +AGENT_DIR="/usr/local/lib/roundtable-agent" +CONF_FILE="/etc/roundtable-agent.conf" +UNIT_FILE="/etc/systemd/system/roundtable-agent.service" + +# ── preflight ──────────────────────────────────────────────────────────────── +[ "$(id -u)" = "0" ] || { echo "must run as root (use sudo)"; exit 1; } +command -v systemctl >/dev/null 2>&1 || { echo "systemd not found — unsupported init system"; exit 1; } +command -v python3 >/dev/null 2>&1 || { echo "python3 not found — install python3 first"; exit 1; } + +# Handle --uninstall +if [ "${1:-}" = "--uninstall" ]; then + systemctl disable --now roundtable-agent.service 2>/dev/null || true + rm -f "$UNIT_FILE" "$CONF_FILE" + rm -rf "$AGENT_DIR" + systemctl daemon-reload + # keep the system user — removing it risks orphaned files elsewhere + echo "uninstalled" + exit 0 +fi + +# ── create system user ─────────────────────────────────────────────────────── +if ! id "$AGENT_USER" >/dev/null 2>&1; then + useradd --system --no-create-home --shell /usr/sbin/nologin "$AGENT_USER" +fi + +# ── drop the agent file ────────────────────────────────────────────────────── +mkdir -p "$AGENT_DIR" +curl -sSL "${ROUNDTABLE_URL}/plugins/host_agent/agent.py" -o "$AGENT_DIR/agent.py" +chmod 0755 "$AGENT_DIR/agent.py" +chown root:root "$AGENT_DIR/agent.py" + +# ── write config ───────────────────────────────────────────────────────────── +cat > "$CONF_FILE" < "$UNIT_FILE" <Host Agent — Registered Hosts + +{% if new_token %} +
+

New token — copy this install command now

+

+ This is the only time the raw token is shown. Rotate if you lose it. +

+
curl -sSL '{{ install_url }}' | sudo sh
+
+{% endif %} + +
+
+
+ + +
+
+ + +
+ +
+
+ +
+ + + + + + + + + {% for item in registrations %} + + + + + + + + {% else %} + + {% endfor %} + +
HostAgent versionDistroLast seenActions
{{ item.host.name if item.host else item.reg.host_id }}{{ item.reg.agent_version or "—" }}{{ item.reg.distro or "—" }}{{ item.reg.last_seen_at.strftime("%Y-%m-%d %H:%M:%S") if item.reg.last_seen_at else "never" }} +
+ +
+
+ +
+
No hosts registered. Add one above.
+
+{% endblock %} diff --git a/host_agent/templates/widget_history.html b/host_agent/templates/widget_history.html new file mode 100644 index 0000000..35e363f --- /dev/null +++ b/host_agent/templates/widget_history.html @@ -0,0 +1,24 @@ +
+

{{ host.name }} — last {{ hours }}h

+ + +
diff --git a/host_agent/templates/widget_table.html b/host_agent/templates/widget_table.html new file mode 100644 index 0000000..21d51d7 --- /dev/null +++ b/host_agent/templates/widget_table.html @@ -0,0 +1,19 @@ + + + + + + {% for r in rows %} + + + + + + + + + {% else %} + + {% endfor %} + +
HostCPU %Mem %Disk % (worst)Load 1mLast seen
{{ r.host.name }}{{ "%.1f"|format(r.cpu_pct) if r.cpu_pct is not none else "—" }}{{ "%.1f"|format(r.mem_used_pct) if r.mem_used_pct is not none else "—" }}{{ "%.1f"|format(r.disk_worst) if r.disk_worst is not none else "—" }}{{ "%.2f"|format(r.load_1m) if r.load_1m is not none else "—" }}{{ r.reg.last_seen_at.strftime("%H:%M:%S") if r.reg.last_seen_at else "never" }}
No hosts with agent data yet.