diff --git a/steward/ansible/routes.py b/steward/ansible/routes.py index 02918a6..61ae3aa 100644 --- a/steward/ansible/routes.py +++ b/steward/ansible/routes.py @@ -1,20 +1,28 @@ # steward/ansible/routes.py from __future__ import annotations -import asyncio -import uuid -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from quart import ( - Blueprint, Response, current_app, render_template, - request, session, + Blueprint, Response, current_app, redirect, render_template, + request, session, url_for, ) from sqlalchemy import select from steward.ansible import executor, sources as src_module from steward.auth.middleware import require_role from steward.models.ansible import AnsibleRun, AnsibleRunStatus +from steward.models.ansible_schedule import AnsibleSchedule from steward.models.users import User, UserRole +INTERVAL_PRESETS = [ + (300, "Every 5 minutes"), + (3600, "Hourly"), + (21600, "Every 6 hours"), + (43200, "Every 12 hours"), + (86400, "Daily"), + (604800, "Weekly"), +] + ansible_bp = Blueprint("ansible", __name__, url_prefix="/ansible") @@ -85,11 +93,34 @@ async def view_playbook(source_name: str, playbook_path: str): ) +def _parse_run_params(form) -> tuple[dict | None, str | None]: + """Build the executor params dict from form fields. Returns (params, error).""" + extra_vars: list[str] = [] + for line in (form.get("extra_vars", "") or "").splitlines(): + line = line.strip() + if not line: + continue + if "=" not in line: + return None, f"Invalid extra var (expected key=value): {line!r}" + extra_vars.append(line) + params: dict = {} + if extra_vars: + params["extra_vars"] = extra_vars + limit = (form.get("limit", "") or "").strip() + if limit: + params["limit"] = limit + tags = (form.get("tags", "") or "").strip() + if tags: + params["tags"] = tags + if "check" in form: + params["check"] = True + return (params or None), None + + @ansible_bp.post("/runs") @require_role(UserRole.operator) async def create_run(): - import json as _json - from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory + from steward.ansible.runner import trigger_run form = await request.form playbook_path = form.get("playbook_path", "").strip() @@ -99,82 +130,22 @@ async def create_run(): if not playbook_path: return "playbook_path is required", 400 - all_sources = _get_sources() - source = next((s for s in all_sources if s["name"] == source_name), None) - if source is None: - return "Source not found", 404 + params_or_none, err = _parse_run_params(form) + if err: + return err, 400 - # Resolve inventory for this scope. - inventory_content: str | None = None - inventory_path: str | None = None - if inventory_scope.startswith("steward:"): - async with current_app.db_sessionmaker() as db: - targets = await fetch_scope_targets(db, inventory_scope) - inventory_content = _json.dumps(generate_inventory(targets)) - elif inventory_scope.startswith("repo:"): - parts = inventory_scope.split(":", 2) - inventory_path = parts[2] if len(parts) == 3 else "" - else: - return "Invalid inventory_scope", 400 - - # Optional run parameters (all passed as argv by the executor — no shell). - extra_vars: list[str] = [] - for line in (form.get("extra_vars", "") or "").splitlines(): - line = line.strip() - if not line: - continue - if "=" not in line: - return f"Invalid extra var (expected key=value): {line!r}", 400 - extra_vars.append(line) - limit = (form.get("limit", "") or "").strip() - tags = (form.get("tags", "") or "").strip() - check = "check" in form - - params: dict = {} - if extra_vars: - params["extra_vars"] = extra_vars - if limit: - params["limit"] = limit - if tags: - params["tags"] = tags - if check: - params["check"] = True - params_or_none = params or None - - run_id = str(uuid.uuid4()) - now = datetime.now(timezone.utc) - - run = AnsibleRun( - id=run_id, - playbook_path=playbook_path, - inventory_path=inventory_path, - inventory_scope=inventory_scope, + run, source, err = await trigger_run( + current_app._get_current_object(), # type: ignore[attr-defined] source_name=source_name, - triggered_by=session["user_id"], - status=AnsibleRunStatus.running, - started_at=now, + playbook_path=playbook_path, + inventory_scope=inventory_scope, params=params_or_none, + triggered_by=session["user_id"], ) - async with current_app.db_sessionmaker() as db: - async with db.begin(): - db.add(run) - - task = asyncio.create_task( - executor.start_run( - current_app._get_current_object(), # type: ignore[attr-defined] - run_id, - playbook_path, - inventory_path or "", - source["path"], - params_or_none, - inventory_content, - ) - ) - task.add_done_callback( - lambda t: t.exception() and current_app.logger.error( - "Ansible run %s raised: %s", run_id, t.exception() - ) - ) + if err == "Source not found": + return err, 404 + if err: + return err, 400 return await render_template( "ansible/run_started.html", @@ -233,3 +204,153 @@ async def run_detail(run_id: str): triggered_label = res.scalar_one_or_none() or "(deleted user)" return await render_template( "ansible/run_detail.html", run=run, triggered_label=triggered_label) + + +# ── Scheduled recurring runs ────────────────────────────────────────────────── + +@ansible_bp.get("/schedules") +@require_role(UserRole.viewer) +async def schedules(): + from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup + + editing_id = request.args.get("edit") + async with current_app.db_sessionmaker() as db: + scheds = (await db.execute( + select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars().all() + groups = (await db.execute( + select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all() + targets = (await db.execute( + select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all() + run_ids = [s.last_run_id for s in scheds if s.last_run_id] + runs = { + r.id: r for r in (await db.execute( + select(AnsibleRun).where(AnsibleRun.id.in_(run_ids)))).scalars().all() + } if run_ids else {} + + source_data = [ + {"name": s["name"], "playbooks": src_module.discover_playbooks(s["path"])} + for s in _get_sources() + ] + all_playbooks = sorted({p for sd in source_data for p in sd["playbooks"]}) + + rows = [] + editing = None + for s in scheds: + rows.append({ + "s": s, + "last_run": runs.get(s.last_run_id) if s.last_run_id else None, + "next_run": (s.last_run_at + timedelta(seconds=s.interval_seconds)) if s.last_run_at else None, + }) + if editing_id and s.id == editing_id: + editing = s + + return await render_template( + "ansible/schedules.html", + rows=rows, groups=groups, targets=targets, + source_data=source_data, all_playbooks=all_playbooks, + editing=editing, interval_presets=INTERVAL_PRESETS, + ) + + +def _schedule_form_fields(form) -> tuple[dict | None, str | None]: + """Validate + extract schedule fields from a form. Returns (fields, error).""" + name = (form.get("name", "") or "").strip() + source_name = (form.get("source_name", "") or "").strip() + playbook_path = (form.get("playbook_path", "") or "").strip() + inventory_scope = (form.get("inventory_scope", "steward:all") or "steward:all").strip() + try: + interval = int(form.get("interval_seconds", "0")) + except (TypeError, ValueError): + interval = 0 + if not name or not source_name or not playbook_path or interval <= 0: + return None, "name, source, playbook, and a positive interval are required" + params, err = _parse_run_params(form) + if err: + return None, err + return { + "name": name, "source_name": source_name, "playbook_path": playbook_path, + "inventory_scope": inventory_scope, "interval_seconds": interval, "params": params, + }, None + + +@ansible_bp.post("/schedules") +@require_role(UserRole.operator) +async def create_schedule(): + form = await request.form + fields, err = _schedule_form_fields(form) + if err: + return err, 400 + async with current_app.db_sessionmaker() as db: + async with db.begin(): + db.add(AnsibleSchedule(enabled=True, **fields)) + return redirect(url_for("ansible.schedules")) + + +@ansible_bp.post("/schedules/") +@require_role(UserRole.operator) +async def update_schedule(schedule_id: str): + form = await request.form + fields, err = _schedule_form_fields(form) + if err: + return err, 400 + async with current_app.db_sessionmaker() as db: + async with db.begin(): + sched = await db.get(AnsibleSchedule, schedule_id) + if sched is None: + return "Not found", 404 + for k, v in fields.items(): + setattr(sched, k, v) + return redirect(url_for("ansible.schedules")) + + +@ansible_bp.post("/schedules//toggle") +@require_role(UserRole.operator) +async def toggle_schedule(schedule_id: str): + async with current_app.db_sessionmaker() as db: + async with db.begin(): + sched = await db.get(AnsibleSchedule, schedule_id) + if sched is None: + return "Not found", 404 + sched.enabled = not sched.enabled + return redirect(url_for("ansible.schedules")) + + +@ansible_bp.post("/schedules//delete") +@require_role(UserRole.operator) +async def delete_schedule(schedule_id: str): + async with current_app.db_sessionmaker() as db: + async with db.begin(): + sched = await db.get(AnsibleSchedule, schedule_id) + if sched is not None: + await db.delete(sched) + return redirect(url_for("ansible.schedules")) + + +@ansible_bp.post("/schedules//run-now") +@require_role(UserRole.operator) +async def run_schedule_now(schedule_id: str): + from steward.ansible.runner import trigger_run + + async with current_app.db_sessionmaker() as db: + sched = await db.get(AnsibleSchedule, schedule_id) + if sched is None: + return "Not found", 404 + source_name, playbook_path = sched.source_name, sched.playbook_path + inventory_scope, params = sched.inventory_scope, sched.params + + run, _source, err = await trigger_run( + current_app._get_current_object(), # type: ignore[attr-defined] + source_name=source_name, playbook_path=playbook_path, + inventory_scope=inventory_scope, params=params, + triggered_by=session["user_id"], + ) + async with current_app.db_sessionmaker() as db: + async with db.begin(): + fresh = await db.get(AnsibleSchedule, schedule_id) + if fresh: + fresh.last_run_at = datetime.now(timezone.utc) + fresh.last_run_id = run.id if run else None + fresh.last_error = err + if run: + return redirect(url_for("ansible.run_detail", run_id=run.id)) + return redirect(url_for("ansible.schedules")) diff --git a/steward/ansible/runner.py b/steward/ansible/runner.py new file mode 100644 index 0000000..1aada6b --- /dev/null +++ b/steward/ansible/runner.py @@ -0,0 +1,78 @@ +# steward/ansible/runner.py +"""Shared playbook-run launcher used by the manual route, alerts, and schedules. + +Centralises the resolve-inventory → create AnsibleRun → launch executor flow so +manual, alert-triggered, and scheduled runs all take the exact same path. +""" +from __future__ import annotations + +import asyncio +import json +import uuid +from datetime import datetime, timezone + +from steward.ansible import executor, sources as src_module +from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory +from steward.models.ansible import AnsibleRun, AnsibleRunStatus + + +async def trigger_run( + app, + *, + source_name: str, + playbook_path: str, + inventory_scope: str = "steward:all", + params: dict | None = None, + triggered_by: str | None = None, +): + """Resolve inventory for the scope, persist an AnsibleRun, and launch it. + + triggered_by=None marks a system/automated run (alerts, schedules). + Returns (run, source, error): on success error is None; on failure run is + None and error is a short human-readable reason. + """ + sources = src_module.get_sources(app.config.get("ANSIBLE", {})) + source = next((s for s in sources if s["name"] == source_name), None) + if source is None: + return None, None, "Source not found" + + inventory_content: str | None = None + inventory_path: str | None = None + if inventory_scope.startswith("steward:"): + async with app.db_sessionmaker() as db: + targets = await fetch_scope_targets(db, inventory_scope) + inventory_content = json.dumps(generate_inventory(targets)) + elif inventory_scope.startswith("repo:"): + parts = inventory_scope.split(":", 2) + inventory_path = parts[2] if len(parts) == 3 else "" + else: + return None, source, "Invalid inventory_scope" + + run_id = str(uuid.uuid4()) + run = AnsibleRun( + id=run_id, + playbook_path=playbook_path, + inventory_path=inventory_path, + inventory_scope=inventory_scope, + source_name=source_name, + triggered_by=triggered_by, + status=AnsibleRunStatus.running, + started_at=datetime.now(timezone.utc), + params=params or None, + ) + async with app.db_sessionmaker() as db: + async with db.begin(): + db.add(run) + + task = asyncio.create_task( + executor.start_run( + app, run_id, playbook_path, inventory_path or "", + source["path"], params or None, inventory_content, + ) + ) + task.add_done_callback( + lambda t: t.exception() and app.logger.error( + "Ansible run %s raised: %s", run_id, t.exception() + ) + ) + return run, source, None diff --git a/steward/ansible/scheduler.py b/steward/ansible/scheduler.py new file mode 100644 index 0000000..f4efaff --- /dev/null +++ b/steward/ansible/scheduler.py @@ -0,0 +1,66 @@ +# steward/ansible/scheduler.py +"""Fire due Ansible schedules. Driven by the core ScheduledTask loop (~60s).""" +from __future__ import annotations + +import logging +from datetime import datetime, timezone + +from sqlalchemy import select + +from steward.ansible.runner import trigger_run +from steward.models.ansible_schedule import AnsibleSchedule + +logger = logging.getLogger(__name__) + + +def is_due(schedule: AnsibleSchedule, now: datetime) -> bool: + """True when an enabled schedule has never run or its interval has elapsed. + + Cadence is measured from the previous *fire* time (last_run_at), not run + completion, so a long-running playbook doesn't cause catch-up storms. + """ + if not schedule.enabled: + return False + if schedule.last_run_at is None: + return True + return (now - schedule.last_run_at).total_seconds() >= schedule.interval_seconds + + +async def run_due_schedules(app) -> int: + """Trigger every due schedule once. Returns the number launched.""" + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as db: + schedules = (await db.execute( + select(AnsibleSchedule).where(AnsibleSchedule.enabled.is_(True)) + )).scalars().all() + + fired = 0 + for s in schedules: + if not is_due(s, now): + continue + run = None + err: str | None = None + try: + run, _source, err = await trigger_run( + app, + source_name=s.source_name, + playbook_path=s.playbook_path, + inventory_scope=s.inventory_scope, + params=s.params, + triggered_by=None, + ) + except Exception: + logger.exception("Schedule %s (%s) failed to launch", s.id, s.name) + err = "launch error" + + async with app.db_sessionmaker() as db: + async with db.begin(): + fresh = await db.get(AnsibleSchedule, s.id) + if fresh: + fresh.last_run_at = now + fresh.last_run_id = run.id if run else None + fresh.last_error = err + if run: + fired += 1 + logger.info("Schedule %s (%s) fired run %s", s.id, s.name, run.id) + return fired diff --git a/steward/app.py b/steward/app.py index f488b87..d5bcf06 100644 --- a/steward/app.py +++ b/steward/app.py @@ -284,6 +284,21 @@ def _register_core_tasks(app: Quart) -> None: ) ) + # Fire due Ansible schedules (recurring playbook runs). Checks every minute; + # each schedule's own interval gates whether it actually runs. + async def run_ansible_schedules(): + from .ansible.scheduler import run_due_schedules + await run_due_schedules(app) + + app._task_registry.append( + ScheduledTask( + name="ansible_scheduled_runs", + coro_factory=run_ansible_schedules, + interval_seconds=60, + run_on_startup=False, + ) + ) + async def _mark_interrupted_runs(app: Quart) -> None: from sqlalchemy import update diff --git a/steward/migrations/versions/0018_ansible_schedules.py b/steward/migrations/versions/0018_ansible_schedules.py new file mode 100644 index 0000000..74e49d5 --- /dev/null +++ b/steward/migrations/versions/0018_ansible_schedules.py @@ -0,0 +1,36 @@ +"""Ansible schedules table — recurring playbook runs + +Revision ID: 0018_ansible_schedules +Revises: 0017_ansible_run_scope +Create Date: 2026-06-16 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0018_ansible_schedules" +down_revision: Union[str, None] = "0017_ansible_run_scope" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "ansible_schedules", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("name", sa.String(128), nullable=False), + sa.Column("source_name", sa.String(128), nullable=False), + sa.Column("playbook_path", sa.String(512), nullable=False), + sa.Column("inventory_scope", sa.String(255), nullable=False, server_default="steward:all"), + sa.Column("params", sa.JSON(), nullable=True), + sa.Column("interval_seconds", sa.Integer(), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.true()), + sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_run_id", sa.String(36), nullable=True), + sa.Column("last_error", sa.String(255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("ansible_schedules") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index d8502b5..5f84866 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -6,6 +6,7 @@ from .metrics import PluginMetric from .alerts import AlertRule, AlertState, AlertEvent, AlertOperator, AlertStateEnum from .ansible import AnsibleRun, AnsibleRunStatus from .ansible_inventory import AnsibleTarget, AnsibleGroup, ansible_target_groups +from .ansible_schedule import AnsibleSchedule from .settings import AppSetting from .dashboard import Dashboard, DashboardWidget, DashboardShareToken @@ -17,6 +18,7 @@ __all__ = [ "AlertRule", "AlertState", "AlertEvent", "AlertOperator", "AlertStateEnum", "AnsibleRun", "AnsibleRunStatus", "AnsibleTarget", "AnsibleGroup", "ansible_target_groups", + "AnsibleSchedule", "AppSetting", "Dashboard", "DashboardWidget", "DashboardShareToken", ] diff --git a/steward/models/ansible_schedule.py b/steward/models/ansible_schedule.py new file mode 100644 index 0000000..9804c68 --- /dev/null +++ b/steward/models/ansible_schedule.py @@ -0,0 +1,34 @@ +from __future__ import annotations +import uuid +from datetime import datetime, timezone +from sqlalchemy import Boolean, DateTime, Integer, JSON, String +from sqlalchemy.orm import Mapped, mapped_column +from .base import Base + + +class AnsibleSchedule(Base): + """A recurring Ansible playbook run. + + Fired by the core scheduler (ansible_scheduled_runs task) on an interval; + each firing creates a system-triggered AnsibleRun (triggered_by=None). The + actual run outcome is read back via last_run_id → AnsibleRun.status, so we + only store provenance here, not a duplicated status. + """ + __tablename__ = "ansible_schedules" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) + name: Mapped[str] = mapped_column(String(128), nullable=False) + source_name: Mapped[str] = mapped_column(String(128), nullable=False) + playbook_path: Mapped[str] = mapped_column(String(512), nullable=False) + inventory_scope: Mapped[str] = mapped_column(String(255), nullable=False, default="steward:all") + # {extra_vars: [...], limit, tags, check} — same shape as AnsibleRun.params. + params: Mapped[dict | None] = mapped_column(JSON, nullable=True) + interval_seconds: Mapped[int] = mapped_column(Integer, nullable=False) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_run_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + # Set only when a firing failed to launch (e.g. source vanished); else NULL. + last_error: Mapped[str | None] = mapped_column(String(255), nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc) + ) diff --git a/steward/templates/ansible/index.html b/steward/templates/ansible/index.html index 9807f46..c77be9a 100644 --- a/steward/templates/ansible/index.html +++ b/steward/templates/ansible/index.html @@ -3,7 +3,10 @@ {% block content %}

Ansible Runs

- Browse Playbooks +
{% if not runs %}
diff --git a/steward/templates/ansible/schedules.html b/steward/templates/ansible/schedules.html new file mode 100644 index 0000000..470fd6f --- /dev/null +++ b/steward/templates/ansible/schedules.html @@ -0,0 +1,141 @@ +{% extends "base.html" %} +{% block title %}Ansible Schedules — Steward{% endblock %} + +{% macro fmt_interval(secs) %} +{%- set m = {300: "every 5 min", 3600: "hourly", 21600: "every 6h", 43200: "every 12h", 86400: "daily", 604800: "weekly"} -%} +{{ m.get(secs, secs ~ "s") }} +{% endmacro %} + +{% block content %} +
+

Ansible Schedules

+ ← Runs +
+ +{# ── Existing schedules ───────────────────────────────────────────────────── #} +{% if not rows %} +
No schedules yet. Create one below to run a playbook on a recurring interval.
+{% else %} +
+ + + + + + + + + {% for row in rows %} + {% set s = row.s %} + + + + + + + + + + {% endfor %} + +
NamePlaybookTargetIntervalLast runNext run
{{ s.name }}{{ s.source_name }} / {{ s.playbook_path }}{{ s.inventory_scope }}{{ fmt_interval(s.interval_seconds) }} + {% if s.last_error %} + launch error + {% elif row.last_run %} + {% set sc = {"running":"var(--yellow)","success":"var(--green)","failed":"var(--red)","interrupted":"var(--orange)"}.get(row.last_run.status.value, "var(--text-muted)") %} + {{ row.last_run.status.value }} + · {{ s.last_run_at.strftime("%m-%d %H:%M") }} + {% else %} + never + {% endif %} + + {% if not s.enabled %}paused + {% elif row.next_run %}{{ row.next_run.strftime("%m-%d %H:%M") }} + {% else %}due now{% endif %} + +
+ +
+
+ +
+ Edit +
+ +
+
+
+{% endif %} + +{# ── Create / edit form ───────────────────────────────────────────────────── #} +{% set p = (editing.params or {}) if editing else {} %} +
+

{{ "Edit schedule" if editing else "New schedule" }}

+
+
+
+ + +
+
+ + +
+
+ + + + {% for pb in all_playbooks %}{% endfor %} + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+
+ + +
+
+ + +
+
+ + {% if editing %}Cancel{% endif %} +
+
+
+{% endblock %} diff --git a/tests/test_ansible_schedule_due.py b/tests/test_ansible_schedule_due.py new file mode 100644 index 0000000..28d6658 --- /dev/null +++ b/tests/test_ansible_schedule_due.py @@ -0,0 +1,28 @@ +"""Unit tests for the schedule due-check (pure, no DB).""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +from steward.ansible.scheduler import is_due + +NOW = datetime(2026, 6, 16, 12, 0, 0, tzinfo=timezone.utc) + + +def _sched(enabled=True, last=None, interval=3600): + return SimpleNamespace(enabled=enabled, last_run_at=last, interval_seconds=interval) + + +def test_never_run_is_due(): + assert is_due(_sched(last=None), NOW) is True + + +def test_disabled_is_never_due(): + assert is_due(_sched(enabled=False, last=None), NOW) is False + + +def test_due_when_interval_elapsed(): + assert is_due(_sched(last=NOW - timedelta(seconds=3600), interval=3600), NOW) is True + assert is_due(_sched(last=NOW - timedelta(seconds=3700), interval=3600), NOW) is True + + +def test_not_due_before_interval(): + assert is_due(_sched(last=NOW - timedelta(seconds=1800), interval=3600), NOW) is False