diff --git a/plugins/prometheus/__init__.py b/plugins/prometheus/__init__.py new file mode 100644 index 0000000..c826d61 --- /dev/null +++ b/plugins/prometheus/__init__.py @@ -0,0 +1,41 @@ +# plugins/prometheus/__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 + from .models import PrometheusTarget # noqa: F401 — registers the table + + +def get_scheduled_tasks() -> list: + from .scheduler import make_scrape_task + return [make_scrape_task(_app)] + + +def get_blueprint(): + from .routes import prometheus_bp + return prometheus_bp + + +def get_nav() -> list[dict]: + # Surfaces in the sidebar's Infrastructure group (scraped endpoints). + return [{"label": "Prometheus", "href": "/plugins/prometheus/"}] + + +def get_alert_metrics() -> dict[str, list[str]]: + """The metric names are whatever each endpoint exposes, so the list is + empty here and the rule form offers the names recorded so far.""" + return {"prometheus": []} + + +def get_audit_entities() -> dict: + """Audit rows about a target name it and link to its page.""" + from .models import PrometheusTarget + return {"prometheus_target": (PrometheusTarget, "/plugins/prometheus/targets/{}")} diff --git a/plugins/prometheus/migrations/__init__.py b/plugins/prometheus/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/prometheus/migrations/versions/__init__.py b/plugins/prometheus/migrations/versions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/plugins/prometheus/migrations/versions/prometheus_001_targets.py b/plugins/prometheus/migrations/versions/prometheus_001_targets.py new file mode 100644 index 0000000..c8358a4 --- /dev/null +++ b/plugins/prometheus/migrations/versions/prometheus_001_targets.py @@ -0,0 +1,46 @@ +"""Prometheus scrape targets are rows edited on the Prometheus page + +Each row is a /metrics endpoint: where it is, its bearer token (encrypted), +how often to scrape it, which series to keep, and the newest scrape's +outcome (milestone 506). + +Revision ID: prometheus_001_targets +Revises: (none — a branch off core via depends_on) +Create Date: 2026-10-12 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "prometheus_001_targets" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = "prometheus" +depends_on: Union[str, Sequence[str], None] = ("0004_app_settings",) + + +def upgrade() -> None: + op.create_table( + "prometheus_targets", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("name", sa.String(100), nullable=False, unique=True), + sa.Column("url", sa.String(500), nullable=False), + sa.Column("token", sa.Text, nullable=False, server_default=""), + sa.Column("verify_tls", sa.Boolean, nullable=False, server_default=sa.true()), + sa.Column("interval_seconds", sa.Integer, nullable=False, server_default="60"), + sa.Column("keep", sa.JSON, nullable=False, server_default="[]"), + sa.Column("drop", sa.JSON, nullable=False, server_default="[]"), + sa.Column("series_cap", sa.Integer, nullable=False, server_default="1000"), + sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.true()), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_scraped_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_ok_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.Text, nullable=True), + sa.Column("last_kept", sa.Integer, nullable=False, server_default="0"), + sa.Column("last_dropped", sa.Integer, nullable=False, server_default="0"), + sa.Column("last_capped", sa.Integer, nullable=False, server_default="0"), + ) + + +def downgrade() -> None: + op.drop_table("prometheus_targets") diff --git a/plugins/prometheus/models.py b/plugins/prometheus/models.py new file mode 100644 index 0000000..d67f53f --- /dev/null +++ b/plugins/prometheus/models.py @@ -0,0 +1,41 @@ +# plugins/prometheus/models.py +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import JSON, Boolean, DateTime, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from steward.core.ids import time_ordered_id +from steward.models.base import Base + + +class PrometheusTarget(Base): + """A /metrics endpoint Steward scrapes, edited on the Prometheus page. + + The bearer token is stored encrypted. keep and drop are lists of name + patterns ("fabledarchives_*") an admin chooses: Steward does the + filtering, the endpoint serves everything. The last_* columns are the + newest scrape's outcome, written by the scraper, so a page can say why a + target isn't answering without asking it. + """ + __tablename__ = "prometheus_targets" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + name: Mapped[str] = mapped_column(String(100), nullable=False, unique=True) + url: Mapped[str] = mapped_column(String(500), nullable=False) + token: Mapped[str] = mapped_column(Text, nullable=False, default="") + verify_tls: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + interval_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=60) + keep: Mapped[list] = mapped_column(JSON, nullable=False, default=list) + drop: Mapped[list] = mapped_column(JSON, nullable=False, default=list) + series_cap: Mapped[int] = mapped_column(Integer, nullable=False, default=1000) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + last_scraped_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_ok_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + last_kept: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + last_dropped: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + last_capped: Mapped[int] = mapped_column(Integer, nullable=False, default=0) diff --git a/plugins/prometheus/plugin.yaml b/plugins/prometheus/plugin.yaml new file mode 100644 index 0000000..ff839d8 --- /dev/null +++ b/plugins/prometheus/plugin.yaml @@ -0,0 +1,17 @@ +# plugins/prometheus/plugin.yaml +name: prometheus +version: "1.0.0" +description: "Scrapes Prometheus /metrics endpoints into Steward's metrics, so any service that exposes them can be charted and alerted on." +author: "Steward" +license: "MIT" +min_app_version: "0.1.0" +repository_url: "https://git.fabledsword.com/bvandeusen/FabledSteward" +homepage: "https://git.fabledsword.com/bvandeusen/FabledSteward/src/branch/dev/plugins/prometheus" +tags: + - prometheus + - metrics + - monitoring + +# Targets are added, tested and edited on the Prometheus page +# (/plugins/prometheus/); each carries its own interval. +config: {} diff --git a/plugins/prometheus/routes.py b/plugins/prometheus/routes.py new file mode 100644 index 0000000..43e3c6b --- /dev/null +++ b/plugins/prometheus/routes.py @@ -0,0 +1,313 @@ +"""The Prometheus pages. Every GET reads Steward's own records: the target +rows the editor writes and the scraper annotates, and the series it stores. +Only Test, an admin's explicit action, asks an endpoint anything (System +178's ruling).""" +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone + +from quart import Blueprint, abort, current_app, redirect, render_template, request, session +from sqlalchemy import delete, func, select, update + +from steward.auth.middleware import require_role +from steward.core.audit import log_audit +from steward.core.crypto import decrypt_secret +from steward.core.formatting import metric_value +from steward.core.metric_series import latest_values, points +from steward.models.metrics import PluginMetric, PluginMetricLatest +from steward.models.users import UserRole + +from .models import PrometheusTarget +from .scraper import ScrapeError, scrape +from .service import (OWN_METRICS, SOURCE, FormError, form_values, map_families, parse_form, + patterns, target_state) + +prometheus_bp = Blueprint("prometheus", __name__, template_folder="templates") + +_REACH_ORDER = {"down": 0, "stale": 1, "pending": 2, "up": 3, "off": 4} + + +def _actor() -> tuple[str | None, str]: + return session.get("user_id"), session.get("username", "") + + +async def _all_targets(db) -> list[PrometheusTarget]: + return list((await db.execute(select(PrometheusTarget).order_by(PrometheusTarget.name))).scalars().all()) + + +def view(target: PrometheusTarget, own: dict, now: datetime) -> dict: + """What a page says about a target: its state in words and the numbers + from its newest scrape.""" + state = target_state(target, now) + return {"id": target.id, "name": target.name, "url": target.url, "reach": state.reach, + "says": state.says, "scraped_at": target.last_scraped_at, "ok_at": target.last_ok_at, + "interval": target.interval_seconds, "kept": target.last_kept, + "dropped": target.last_dropped, "capped": target.last_capped, "cap": target.series_cap, + "keep": list(target.keep or []), "drop": list(target.drop or []), + "scrape_ms": own.get("scrape_ms")} + + +def shown_value(metric: str, value: float | None) -> dict: + """A series' value in its unit, from the Prometheus naming conventions. + A *_timestamp_seconds is a moment: it comes back as `moment` for the page + to show as "3m ago" in the reader's zone; 0 there means "never".""" + if value is None: + return {"text": "—", "moment": None} + if metric.endswith("_timestamp_seconds"): + if value <= 0: + return {"text": "never", "moment": None} + return {"text": "", "moment": datetime.fromtimestamp(value, timezone.utc)} + return {"text": metric_value(metric, value), "moment": None} + + +@prometheus_bp.get("/") +@require_role(UserRole.viewer) +async def index(): + async with current_app.db_sessionmaker() as db: + targets = await _all_targets(db) + own = {} + if targets: + rows = (await db.execute(select(PluginMetricLatest).where( + PluginMetricLatest.source_module == SOURCE, + PluginMetricLatest.resource_name.in_([t.name for t in targets]), + PluginMetricLatest.metric_name == "scrape_ms"))).scalars() + own = {r.resource_name: {"scrape_ms": r.value} for r in rows} + now = datetime.now(timezone.utc) + shown = [view(t, own.get(t.name, {}), now) for t in targets] + shown.sort(key=lambda v: (_REACH_ORDER[v["reach"]], v["name"].lower())) + return await render_template("prometheus/index.html", targets=shown) + + +# ── The editor ──────────────────────────────────────────────────────────────── + +async def _render_form(target: PrometheusTarget | None, values: dict, error: str | None = None, + status: int = 200): + return await render_template("prometheus/form.html", target=target, v=values, error=error), status + + +def _typed(form, target: PrometheusTarget | None = None) -> dict: + """What was typed, to show the form again with an error. The token is not + echoed back.""" + return {"name": form.get("name", ""), "url": form.get("url", ""), + "interval_seconds": form.get("interval_seconds", ""), "series_cap": form.get("series_cap", ""), + "keep": form.get("keep", ""), "drop": form.get("drop", ""), + "verify_tls": bool(form.get("verify_tls")), "enabled": bool(form.get("enabled")), + "has_token": bool(target and target.token)} + + +async def _name_taken(db, name: str, other_than: str | None = None) -> bool: + q = select(PrometheusTarget.id).where(PrometheusTarget.name == name) + if other_than: + q = q.where(PrometheusTarget.id != other_than) + return (await db.execute(q)).first() is not None + + +async def _target_or_404(db, target_id: str) -> PrometheusTarget: + target = await db.get(PrometheusTarget, target_id) + if target is None: + abort(404) + return target + + +@prometheus_bp.get("/targets/new") +@require_role(UserRole.admin) +async def new_target(): + return await _render_form(None, form_values(None)) + + +@prometheus_bp.post("/targets/new") +@require_role(UserRole.admin) +async def create_target(): + form = await request.form + try: + values = parse_form(form) + except FormError as exc: + return await _render_form(None, _typed(form), str(exc), 422) + async with current_app.db_sessionmaker() as db: + async with db.begin(): + if await _name_taken(db, values["name"]): + return await _render_form(None, _typed(form), + f"A target is already named “{values['name']}”.", 422) + target = PrometheusTarget(**values) + db.add(target) + await db.flush() + target_id = target.id + await log_audit(current_app, *_actor(), "prometheus.target_created", entity_type="prometheus_target", + entity_id=target_id, detail={"name": values["name"], "url": values["url"]}) + return redirect(f"/plugins/prometheus/targets/{target_id}") + + +@prometheus_bp.get("/targets//edit") +@require_role(UserRole.admin) +async def edit_target(target_id: str): + async with current_app.db_sessionmaker() as db: + target = await _target_or_404(db, target_id) + return await _render_form(target, form_values(target)) + + +@prometheus_bp.post("/targets//edit") +@require_role(UserRole.admin) +async def save_target(target_id: str): + form = await request.form + async with current_app.db_sessionmaker() as db: + async with db.begin(): + target = await _target_or_404(db, target_id) + try: + values = parse_form(form, target) + except FormError as exc: + return await _render_form(target, _typed(form, target), str(exc), 422) + if await _name_taken(db, values["name"], other_than=target.id): + return await _render_form(target, _typed(form, target), + f"A target is already named “{values['name']}”.", 422) + old_name = target.name + for column, value in values.items(): + setattr(target, column, value) + # Where it is or what is kept may have changed: the last outcome no + # longer describes it, so the page says "not scraped yet" until the next scrape. + target.last_scraped_at = target.last_ok_at = target.last_error = None + target.last_kept = target.last_dropped = target.last_capped = 0 + if old_name != target.name: + # Its history is stored under its name, and its labelled series + # under ":…": move them with the rename. + for model in (PluginMetric, PluginMetricLatest): + await db.execute(update(model).where( + model.source_module == SOURCE, model.resource_name == old_name + ).values(resource_name=target.name)) + # ":labels" keeps its ":labels" (substr is 1-based). + await db.execute(update(model).where( + model.source_module == SOURCE, + model.resource_name.startswith(old_name + ":", autoescape=True), + ).values(resource_name=target.name + func.substr(model.resource_name, len(old_name) + 1))) + from . import scheduler + scheduler.RATES.forget(target_id) + detail = {"name": values["name"]} + if old_name != values["name"]: + detail["old_name"] = old_name + await log_audit(current_app, *_actor(), "prometheus.target_updated", entity_type="prometheus_target", + entity_id=target_id, detail=detail) + return redirect(f"/plugins/prometheus/targets/{target_id}") + + +@prometheus_bp.post("/targets//delete") +@require_role(UserRole.admin) +async def delete_target(target_id: str): + async with current_app.db_sessionmaker() as db: + async with db.begin(): + target = await _target_or_404(db, target_id) + name = target.name + await db.delete(target) + # Its newest values go with it, so nothing reads it as a target + # that stopped answering. Its history ages out with retention. + await db.execute(delete(PluginMetricLatest).where( + PluginMetricLatest.source_module == SOURCE, + (PluginMetricLatest.resource_name == name) + | PluginMetricLatest.resource_name.startswith(name + ":", autoescape=True))) + from . import scheduler + scheduler.RATES.forget(target_id) + await log_audit(current_app, *_actor(), "prometheus.target_deleted", entity_type="prometheus_target", + entity_id=target_id, detail={"name": name}) + return redirect("/plugins/prometheus/") + + +TEST_FAMILIES_SHOWN = 40 + + +@prometheus_bp.post("/targets/test") +@require_role(UserRole.admin) +async def test_target(): + """Scrape the (unsaved) form's address once and say what would be stored. + An admin's explicit action, so it may contact the endpoint.""" + form = await request.form + existing = None + if form.get("target_id"): + async with current_app.db_sessionmaker() as db: + existing = await db.get(PrometheusTarget, form["target_id"]) + url = (form.get("url") or "").strip() + token = (form.get("token") or "").strip() + if not token and existing is not None and existing.token and not form.get("clear_token"): + token = decrypt_secret(existing.token, context=f"prometheus target {existing.name} token") + found: dict = {} + if not url.startswith(("http://", "https://")): + found["error"] = "Enter the address first: http://… or https://…" + else: + try: + families = await scrape(url, token, bool(form.get("verify_tls"))) + except ScrapeError as exc: + found["error"] = str(exc) + else: + try: + cap = max(1, int(form.get("series_cap") or 1000)) + except ValueError: + cap = 1000 + keep, drop = patterns(form.get("keep") or ""), patterns(form.get("drop") or "") + mapped = map_families("t", families, keep, drop, cap) + kept_names = {} + for s in mapped.series: + kept_names[s.metric] = kept_names.get(s.metric, 0) + 1 + found = {"families": len(families), "kept": len(mapped.series), "dropped": mapped.dropped, + "capped": mapped.capped, + "names": sorted(kept_names.items())[:TEST_FAMILIES_SHOWN], + "more": max(0, len(kept_names) - TEST_FAMILIES_SHOWN)} + return await render_template("prometheus/_test.html", found=found, url=url) + + +# ── The target page and one series ─────────────────────────────────────────── + +def group_series(target_name: str, latest: list) -> list[dict]: + """The target's stored series grouped by metric, each with its rows: + the labels that tell them apart, and the newest value.""" + groups: dict[str, list] = {} + for row in latest: + if row.resource_name == target_name and row.metric_name in OWN_METRICS: + continue + labels = row.resource_name[len(target_name) + 1:] if row.resource_name != target_name else "" + groups.setdefault(row.metric_name, []).append({ + "labels": labels, "resource": row.resource_name, "at": row.recorded_at, + **shown_value(row.metric_name, row.value)}) + return [{"metric": m, "rows": sorted(rows, key=lambda r: r["labels"])} + for m, rows in sorted(groups.items())] + + +@prometheus_bp.get("/targets/") +@require_role(UserRole.viewer) +async def target_detail(target_id: str): + q = (request.args.get("q") or "").strip() + async with current_app.db_sessionmaker() as db: + target = await _target_or_404(db, target_id) + latest = await latest_values(db, SOURCE, target.name, include_children=True) + own = {r.metric_name: r.value for r in latest if r.resource_name == target.name + and r.metric_name in OWN_METRICS} + groups = group_series(target.name, latest) + total = sum(len(g["rows"]) for g in groups) + if q: + groups = [g for g in groups if q.lower() in g["metric"].lower()] + now = datetime.now(timezone.utc) + return await render_template("prometheus/target.html", target=view(target, own, now), groups=groups, + total=total, q=q) + + +RANGES = (1, 6, 24, 168) + + +@prometheus_bp.get("/targets//series") +@require_role(UserRole.viewer) +async def series_detail(target_id: str): + resource = request.args.get("resource") or "" + metric = request.args.get("metric") or "" + hours = request.args.get("hours", type=int, default=24) + if hours not in RANGES: + hours = 24 + async with current_app.db_sessionmaker() as db: + target = await _target_or_404(db, target_id) + if resource != target.name and not resource.startswith(target.name + ":"): + abort(404) + now_row = await db.get(PluginMetricLatest, (SOURCE, resource, metric)) + series = await points(db, SOURCE, resource, metric, + datetime.now(timezone.utc) - timedelta(hours=hours)) + labels = resource[len(target.name) + 1:] if resource != target.name else "" + stamp_axis = metric.endswith("_timestamp_seconds") + return await render_template( + "prometheus/series.html", target=target, resource=resource, metric=metric, labels=labels, + now=shown_value(metric, now_row.value if now_row else None), hours=hours, ranges=RANGES, + points_json=json.dumps([(t.isoformat(), v) for t, v in series]), stamp_axis=stamp_axis) diff --git a/plugins/prometheus/scheduler.py b/plugins/prometheus/scheduler.py new file mode 100644 index 0000000..8f05b4f --- /dev/null +++ b/plugins/prometheus/scheduler.py @@ -0,0 +1,130 @@ +# plugins/prometheus/scheduler.py +from __future__ import annotations + +import asyncio +import logging +import time +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +from sqlalchemy import select + +from steward.core.counter_rates import CounterRates +from steward.core.scheduler import ScheduledTask + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + +# How often the scheduler looks for due targets. A target is scraped within +# one tick of its interval, so this is the cadence's worst-case lateness. +TICK_SECONDS = 5 + +# A scrape past this is given up, whatever the client's own timeouts did: +# a slow drip of bytes can keep each read inside its timeout for ever. +SCRAPE_DEADLINE_SECONDS = 30 + +# Counters' previous samples, for their rates. +RATES = CounterRates() + + +def make_scrape_task(app: "Quart") -> ScheduledTask: + async def tick() -> None: + await scrape_due(app) + + return ScheduledTask(name="prometheus_scrape", coro_factory=tick, + interval_seconds=TICK_SECONDS, run_on_startup=True) + + +def is_due(target, now: datetime) -> bool: + """Whether a target's interval has elapsed since its last scrape, with + half a tick of slack so tick jitter doesn't push it back a whole tick.""" + if not target.enabled: + return False + if target.last_scraped_at is None: + return True + elapsed = (now - target.last_scraped_at).total_seconds() + return elapsed >= target.interval_seconds - TICK_SECONDS / 2 + + +async def scrape_due(app: "Quart", scrape=None) -> None: + """Scrape every due target at once, then write each one's outcome to its + row and its series to the metric tables. `scrape` stands in for + scraper.scrape in tests.""" + from steward.core.crypto import decrypt_secret + + from .models import PrometheusTarget + from .scraper import ScrapeError + from .scraper import scrape as fetch + + scrape = scrape or fetch + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as session: + targets = [t for t in (await session.execute(select(PrometheusTarget))).scalars().all() + if is_due(t, now)] + if not targets: + return + + async def one(t): + started = time.monotonic() + token = decrypt_secret(t.token, context=f"prometheus target {t.name} token") + try: + families = await asyncio.wait_for(scrape(t.url, token, t.verify_tls), + SCRAPE_DEADLINE_SECONDS) + return families, (time.monotonic() - started) * 1000 + except TimeoutError: + return ScrapeError(f"Gave up after {SCRAPE_DEADLINE_SECONDS} s."), \ + (time.monotonic() - started) * 1000 + except Exception as exc: # noqa: BLE001 — every failure is the target's outcome + return exc, (time.monotonic() - started) * 1000 + + results = await asyncio.gather(*(one(t) for t in targets)) + await record(app, targets, results, datetime.now(timezone.utc)) + + +async def record(app: "Quart", targets, results, now: datetime) -> None: + """Each target's outcome on its row, and its kept series (counters as + rates) plus its own is_up / scrape_ms / series counts as metrics.""" + from steward.core.alerts import MetricBatch + + from .models import PrometheusTarget + from .scraper import ScrapeError + from .service import (IS_UP, SCRAPE_MS, SERIES_CAPPED, SERIES_KEPT, SOURCE, map_families) + + async with app.db_sessionmaker() as session: + async with session.begin(): + batch = MetricBatch() + for target, (result, ms) in zip(targets, results): + row = await session.get(PrometheusTarget, target.id) + if row is None: # deleted while it was being scraped + continue + row.last_scraped_at = now + if isinstance(result, BaseException): + if not isinstance(result, ScrapeError): + logger.warning("Prometheus scrape of %s failed: %r", target.name, result) + row.last_error = str(result) or "The scrape failed." + batch.add(source_module=SOURCE, resource_name=row.name, metric_name=IS_UP, + value=0.0, recorded_at=now) + continue + mapped = map_families(row.name, result, list(row.keep or []), list(row.drop or []), + row.series_cap) + row.last_error = None + row.last_ok_at = now + row.last_kept = len(mapped.series) + row.last_dropped = mapped.dropped + row.last_capped = mapped.capped + for series in mapped.series: + value = series.value + if series.counter: + value = RATES.rate((row.id, series.resource, series.metric), value, now) + if value is None: + continue + batch.add(source_module=SOURCE, resource_name=series.resource, + metric_name=series.metric, value=value, recorded_at=now) + for metric_name, value in [(IS_UP, 1.0), (SCRAPE_MS, ms), + (SERIES_KEPT, len(mapped.series)), + (SERIES_CAPPED, mapped.capped)]: + batch.add(source_module=SOURCE, resource_name=row.name, + metric_name=metric_name, value=value, recorded_at=now) + await batch.record(session) diff --git a/plugins/prometheus/scraper.py b/plugins/prometheus/scraper.py new file mode 100644 index 0000000..29cb0da --- /dev/null +++ b/plugins/prometheus/scraper.py @@ -0,0 +1,58 @@ +# plugins/prometheus/scraper.py +"""Fetching one /metrics endpoint. Every failure comes back as one sentence +a page can show, never a stack trace.""" +from __future__ import annotations + +import httpx + +from steward.core import prometheus + +TIMEOUT_SECONDS = 10.0 +# A /metrics page past this is not one to store a series of; reading stops. +MAX_BYTES = 20 * 1024 * 1024 + +ACCEPT = "text/plain;version=0.0.4;q=1,*/*;q=0.1" + + +class ScrapeError(Exception): + """Why a scrape got nothing, said to the person.""" + + +async def scrape(url: str, token: str = "", verify_tls: bool = True, + transport: httpx.AsyncBaseTransport | None = None) -> dict[str, prometheus.Family]: + """The endpoint's families. `transport` stands in for the network in tests.""" + headers = {"Accept": ACCEPT} + if token: + headers["Authorization"] = f"Bearer {token}" + try: + async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, verify=verify_tls, + transport=transport, follow_redirects=False) as client: + async with client.stream("GET", url, headers=headers) as response: + if response.status_code in (401, 403): + raise ScrapeError(f"Answered {response.status_code}: the token was refused. " + "Check the token, or make a new one on the service.") + if response.status_code != 200: + raise ScrapeError(f"Answered {response.status_code} {response.reason_phrase}, not metrics.") + body = bytearray() + async for chunk in response.aiter_bytes(): + body.extend(chunk) + if len(body) > MAX_BYTES: + raise ScrapeError(f"The page is over {MAX_BYTES // (1024 * 1024)} MB; " + "is this a /metrics address?") + except ScrapeError: + raise + except httpx.TimeoutException: + raise ScrapeError(f"No answer within {TIMEOUT_SECONDS:.0f} s.") from None + except httpx.ConnectError as exc: + raise ScrapeError(f"Couldn't connect: {_reason(exc)}.") from None + except httpx.HTTPError as exc: + raise ScrapeError(f"The request failed: {_reason(exc)}.") from None + families = prometheus.parse(body.decode("utf-8", errors="replace")) + if not families: + raise ScrapeError("It answered, but with no metrics in Prometheus format.") + return families + + +def _reason(exc: Exception) -> str: + text = str(exc).strip() or type(exc).__name__ + return text.rstrip(".") diff --git a/plugins/prometheus/service.py b/plugins/prometheus/service.py new file mode 100644 index 0000000..6edace2 --- /dev/null +++ b/plugins/prometheus/service.py @@ -0,0 +1,194 @@ +"""Prometheus targets: the form a person edits, the row it becomes, how a +scrape's series become Steward metrics, and what a page says about the +newest scrape.""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from datetime import datetime +from fnmatch import fnmatchcase + +from steward.core import prometheus +from steward.core.crypto import encrypt_secret + +from .models import PrometheusTarget + +SOURCE = "prometheus" + +# What each target records about its own scrape, under its own name. +IS_UP = "is_up" +SCRAPE_MS = "scrape_ms" +SERIES_KEPT = "series_kept" +SERIES_CAPPED = "series_capped" +OWN_METRICS = (IS_UP, SCRAPE_MS, SERIES_KEPT, SERIES_CAPPED) + +# Column widths of plugin_metrics; a longer name can't be stored. +RESOURCE_MAX = 255 +METRIC_MAX = 128 + +INTERVAL_MIN, INTERVAL_MAX = 10, 3600 +CAP_MIN, CAP_MAX = 1, 10000 + + +class FormError(ValueError): + """What is wrong with a submitted target form, said to the person.""" + + +# ── Series → metrics ───────────────────────────────────────────────────────── + +def wanted(name: str, keep: list[str], drop: list[str]) -> bool: + """Whether a family is stored: it matches a keep pattern (or there are + none) and no drop pattern. Drop wins over keep.""" + if keep and not any(fnmatchcase(name, p) for p in keep): + return False + return not any(fnmatchcase(name, p) for p in drop) + + +def resource_for(target: str, labels: tuple[tuple[str, str], ...]) -> str: + """The resource a series is stored under: the target's name, then its + labels as key=value, so "Archives:target=Steward,state=late". A series + without labels is the target itself.""" + if not labels: + return target + return target + ":" + ",".join(f"{k}={v}" for k, v in labels) + + +def rate_name(sample_name: str) -> str: + """A counter is stored as a per-second rate: requests_total → + requests_per_sec, job_seconds_sum → job_seconds_sum_per_sec.""" + base = sample_name[: -len("_total")] if sample_name.endswith("_total") else sample_name + return base + "_per_sec" + + +@dataclass +class Series: + resource: str + metric: str + value: float + counter: bool + + +@dataclass +class Mapped: + series: list[Series] = field(default_factory=list) + dropped: int = 0 # left out by the keep/drop patterns + capped: int = 0 # past the series cap, or a name too long to store + + +def map_families(target: str, families: dict[str, prometheus.Family], keep: list[str], + drop: list[str], cap: int) -> Mapped: + """The series of one scrape that Steward stores, in the endpoint's order. + + Gauges and untyped samples are stored as they are; counters (and a + histogram's or summary's _sum and _count) are marked to become rates. + Histogram buckets and summary quantiles are skipped: Steward stores one + number per series, not a distribution. NaN and infinite values are + skipped as unmeasured.""" + out = Mapped() + for family in families.values(): + if not wanted(family.name, keep, drop): + out.dropped += len(family.samples) + continue + for sample in family.samples: + counter = family.type == "counter" + if family.type in ("histogram", "summary"): + if not sample.name.endswith(("_sum", "_count")): + continue + counter = True + if math.isnan(sample.value) or math.isinf(sample.value): + continue + metric = rate_name(sample.name) if counter else sample.name + resource = resource_for(target, sample.labels) + if len(out.series) >= cap or len(metric) > METRIC_MAX or len(resource) > RESOURCE_MAX: + out.capped += 1 + continue + out.series.append(Series(resource, metric, sample.value, counter)) + return out + + +# ── The form ───────────────────────────────────────────────────────────────── + +def patterns(text: str) -> list[str]: + """Name patterns from a textarea: one per line or comma-separated.""" + out: list[str] = [] + for line in (text or "").replace(",", "\n").splitlines(): + p = line.strip() + if p and p not in out: + out.append(p) + return out + + +def form_values(target: PrometheusTarget | None) -> dict: + """The form's fields for an existing target. The token is never sent back + to the browser; a blank token field on save keeps the stored one.""" + if target is None: + return {"interval_seconds": 60, "series_cap": 1000, "verify_tls": True, "enabled": True, + "keep": "", "drop": ""} + return {"name": target.name, "url": target.url, "interval_seconds": target.interval_seconds, + "series_cap": target.series_cap, "verify_tls": target.verify_tls, + "enabled": target.enabled, "keep": "\n".join(target.keep or []), + "drop": "\n".join(target.drop or []), "has_token": bool(target.token)} + + +def _bounded(form, key: str, default: int, low: int, high: int, says: str) -> int: + try: + value = int(form.get(key) or default) + except ValueError: + raise FormError(says) from None + if not low <= value <= high: + raise FormError(says) + return value + + +def parse_form(form, existing: PrometheusTarget | None = None) -> dict: + """The submitted form as the columns to store, the token encrypted. + Raises FormError saying what to fix.""" + name = (form.get("name") or "").strip() + url = (form.get("url") or "").strip() + if not name: + raise FormError("Give the target a name.") + if len(name) > 100 or any(c in name for c in "/:,"): + raise FormError("A name is up to 100 characters, without / : or a comma.") + if not url.startswith(("http://", "https://")) or len(url) > 500: + raise FormError("The address starts with http:// or https://, e.g. http://archives:8081/metrics.") + interval = _bounded(form, "interval_seconds", 60, INTERVAL_MIN, INTERVAL_MAX, + f"Scrape every {INTERVAL_MIN} to {INTERVAL_MAX} seconds.") + cap = _bounded(form, "series_cap", 1000, CAP_MIN, CAP_MAX, + f"The series cap is between {CAP_MIN} and {CAP_MAX}.") + remove_token = bool(form.get("clear_token")) + typed = (form.get("token") or "").strip() + if typed: + token = encrypt_secret(typed) + elif existing is not None and not remove_token: + token = existing.token + else: + token = "" + return {"name": name, "url": url, "token": token, + "verify_tls": bool(form.get("verify_tls")), "interval_seconds": interval, + "keep": patterns(form.get("keep") or ""), "drop": patterns(form.get("drop") or ""), + "series_cap": cap, "enabled": bool(form.get("enabled"))} + + +# ── What the newest scrape says ────────────────────────────────────────────── + +# A target not scraped for this many intervals has stopped being scraped +# (plugin turned off, scheduler stuck), whatever its last outcome said. +STALE_SCRAPES = 3 + + +@dataclass +class TargetState: + reach: str # up | down | stale | pending | off + says: str # one line, in words + + +def target_state(target: PrometheusTarget, now: datetime) -> TargetState: + if not target.enabled: + return TargetState("off", "Not scraped: turned off.") + if target.last_scraped_at is None: + return TargetState("pending", "Not scraped yet.") + if (now - target.last_scraped_at).total_seconds() > STALE_SCRAPES * max(target.interval_seconds, 1): + return TargetState("stale", "Not scraped recently: the Prometheus plugin may be turned off.") + if target.last_error: + return TargetState("down", target.last_error) + return TargetState("up", "Answering.") diff --git a/plugins/prometheus/templates/prometheus/_test.html b/plugins/prometheus/templates/prometheus/_test.html new file mode 100644 index 0000000..b5bcafe --- /dev/null +++ b/plugins/prometheus/templates/prometheus/_test.html @@ -0,0 +1,18 @@ +{# What Test found (milestone 506): how many series the address hands over + and which would be stored with the filters typed, or why it got nothing. #} +{% if found.error %} + +{% else %} +
+

{{ url }} answered with {{ found.families }} metric famil{{ "y" if found.families == 1 else "ies" }}.

+

With what's typed here, {{ found.kept }} series would be stored + {%- if found.dropped %}, {{ found.dropped }} left out by the filters{% endif %} + {%- if found.capped %}, and {{ found.capped }} over the cap{% endif %}.

+ {% if found.names %} +
+ {% for name, n in found.names %}{{ name }}{% if n > 1 %} ×{{ n }}{% endif %}{% endfor %} +
+ {% if found.more %}

and {{ found.more }} more.

{% endif %} + {% endif %} +
+{% endif %} diff --git a/plugins/prometheus/templates/prometheus/form.html b/plugins/prometheus/templates/prometheus/form.html new file mode 100644 index 0000000..adb22ce --- /dev/null +++ b/plugins/prometheus/templates/prometheus/form.html @@ -0,0 +1,111 @@ +{# plugins/prometheus/templates/prometheus/form.html — add or edit a scrape + target (milestone 506): where its /metrics is, its token, how often, and + what to keep, with Test to scrape it once before saving. The token is + never sent back to the browser; a blank token field keeps the stored one. #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% block title %}{{ ("Edit " ~ target.name) if target else "Add Prometheus target" }} — Steward{% endblock %} +{% block breadcrumb %} +{%- if target -%} +{{ crumbs([("Prometheus", "/plugins/prometheus/"), (target.name, "/plugins/prometheus/targets/" ~ target.id), ("Edit", "")]) }} +{%- else -%} +{{ crumbs([("Prometheus", "/plugins/prometheus/"), ("Add target", "")]) }} +{%- endif -%} +{% endblock %} +{% block content %} +{{ page_header(("Edit " ~ target.name) if target else "Add Prometheus target") }} + +
+{% if error %}{% endif %} +
{{ csrf_field() }} + {% if target %}{% endif %} + +
+

Endpoint

+
+
+ + + How its series are named in Steward: alert rules and charts show it. +
+
+ + +
+
+ + + Sent as "Authorization: Bearer …". Blank {{ "keeps the saved one" if v.has_token else "sends none" }}. + {% if v.has_token %} + + {% endif %} +
+
+ + +
+
+
+ + +
+
+ + Scraping… + Scrapes the address once with what's typed here, filters included, and says what would be stored. Nothing is saved. +
+
+
+ +
+

What to keep

+

The endpoint hands over everything; Steward stores what you choose. Patterns match the endpoint's metric names, with * for any run of characters.

+
+
+ + + One a line. Empty keeps everything. +
+
+ + + One a line. Wins over Keep. +
+
+ + + The most series stored from one scrape. Past it, the rest are counted and shown, not stored. +
+
+

Gauges are stored as they are. Counters are stored as a rate per second, named …_per_sec, from the second scrape on. Histogram buckets and summary quantiles aren't stored; their _sum and _count are, as rates.

+
+ +
+ + Cancel +
+
+ +{% if target %} +
+
+

Delete this target

+

Steward stops scraping it. Its past series age out with the rest of the history.

+
+
{{ csrf_field() }} + +
+
+{% endif %} +
+{% endblock %} diff --git a/plugins/prometheus/templates/prometheus/index.html b/plugins/prometheus/templates/prometheus/index.html new file mode 100644 index 0000000..15a126e --- /dev/null +++ b/plugins/prometheus/templates/prometheus/index.html @@ -0,0 +1,59 @@ +{# plugins/prometheus/templates/prometheus/index.html — every scrape target, + those not answering first, each saying what its newest scrape found: how + many series it stored, or why it got nothing (milestone 506). #} +{% extends "base.html" %} +{% from "_macros.html" import page_header, status_badge, stat_tile, empty_state %} +{% block title %}Prometheus — Steward{% endblock %} +{% block content %} +{% set admin = session.user_role == 'admin' %} +{% call page_header("Prometheus targets", (targets|length ~ " target" ~ ("s" if targets|length != 1 else "")) if targets else none) %} + {% if admin %}Add target{% endif %} +{% endcall %} + +{% if not targets %} +
{{ empty_state("No targets yet. Add a service's /metrics address, and Steward scrapes it on a schedule so its numbers can be charted and alerted on.", + "Add target" if admin else none, "/plugins/prometheus/targets/new") }}
+{% endif %} + +{% for t in targets %} +
+
+ {% if t.reach == 'up' %}{{ status_badge('up', 'Answering') }} + {% elif t.reach == 'down' %}{{ status_badge('down', 'Not answering') }} + {% elif t.reach == 'stale' %}{{ status_badge('stale', 'Not scraped') }} + {% elif t.reach == 'off' %}{{ status_badge('stale', 'Turned off') }} + {% else %}{{ status_badge('pending', 'Not scraped yet') }}{% endif %} +

{{ t.name }}

+ {{ t.url }} + + Series + {% if admin %}Edit{% endif %} + +
+ + {% if t.reach == 'down' %} +

{{ t.says }}

+

+ {%- if t.ok_at %}Last answered {{ t.ok_at|ago }}.{% else %}It has never answered.{% endif %} + Tried {{ t.scraped_at|ago }}.

+ {% elif t.reach in ('stale', 'off') %} +

{{ t.says }}{% if t.scraped_at %} Last scraped {{ t.scraped_at|ago }}.{% endif %}

+ {% elif t.reach == 'pending' %} +

Not scraped yet: the first scrape runs within seconds of saving.

+ {% endif %} + + {% if t.reach == 'up' %} +
+ {{ stat_tile("Series stored", t.kept) }} + {{ stat_tile("Over the cap", t.capped, hint=("cap is " ~ t.cap) if t.capped else none) }} + {{ stat_tile("Left out by filters", t.dropped) }} + {{ stat_tile("Scrape time", (t.scrape_ms|round|int ~ " ms") if t.scrape_ms is not none else "—") }} + {{ stat_tile("Every", t.interval|duration) }} +
+ {% if t.capped %} +

{{ t.capped }} series weren't stored: past the cap of {{ t.cap }}, or a name too long to store. {% if admin %}Narrow what's kept or raise the cap.{% endif %}

+ {% endif %} + {% endif %} +
+{% endfor %} +{% endblock %} diff --git a/plugins/prometheus/templates/prometheus/series.html b/plugins/prometheus/templates/prometheus/series.html new file mode 100644 index 0000000..cbd8eca --- /dev/null +++ b/plugins/prometheus/templates/prometheus/series.html @@ -0,0 +1,66 @@ +{# plugins/prometheus/templates/prometheus/series.html — one stored series + of a target (milestone 506): its newest value and a chart over a chosen + range. A *_timestamp_seconds series charts as moments, not numbers. #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% block title %}{{ metric }} — {{ target.name }} — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Prometheus", "/plugins/prometheus/"), (target.name, "/plugins/prometheus/targets/" ~ target.id), (metric, "")]) }}{% endblock %} +{% block content %} +{% call page_header(metric, labels or none) %} + {% for h in ranges %} + {% if h < 168 %}{{ h }}h{% else %}7d{% endif %} + {% endfor %} +{% endcall %} + +

Now: {% if now.moment %}{{ now.moment|ago }}{% else %}{{ now.text }}{% endif %}

+ +
+
+ +
+
+

To be told when it crosses a line, add an alert rule on Prometheus targets → {{ resource }} → {{ metric }}. + New alert rule

+ + +{% endblock %} diff --git a/plugins/prometheus/templates/prometheus/target.html b/plugins/prometheus/templates/prometheus/target.html new file mode 100644 index 0000000..d11bc73 --- /dev/null +++ b/plugins/prometheus/templates/prometheus/target.html @@ -0,0 +1,69 @@ +{# plugins/prometheus/templates/prometheus/target.html — one scrape target + (milestone 506): what its newest scrape found, then every series it + stored, grouped by metric, each with its labels and newest value and a + link to its chart. #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header, status_badge, stat_tile %} +{% block title %}Prometheus — {{ target.name }} — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Prometheus", "/plugins/prometheus/"), (target.name, "")]) }}{% endblock %} +{% block content %} +{% set admin = session.user_role == 'admin' %} +{% call page_header(target.name, target.url) %} + {% if admin %}Edit{% endif %} +{% endcall %} + +

+ {% if target.reach == 'up' %}{{ status_badge('up', 'Answering') }} scraped {{ target.scraped_at|ago }}, every {{ target.interval|duration }} + {% elif target.reach == 'down' %}{{ status_badge('down', 'Not answering') }} {{ target.says }} + {% if target.ok_at %}Last answered {{ target.ok_at|ago }}; the values below are from then.{% else %}It has never answered.{% endif %} + {% elif target.reach == 'stale' %}{{ status_badge('stale', 'Not scraped') }} {{ target.says }} + {% elif target.reach == 'off' %}{{ status_badge('stale', 'Turned off') }} {{ target.says }} + {% else %}{{ status_badge('pending', 'Not scraped yet') }} The first scrape runs within seconds of saving.{% endif %} +

+{% if target.capped %} +

{{ target.capped }} series weren't stored on the last scrape: past the cap of {{ target.cap }}, or a name too long to store. + {% if admin %}Narrow what's kept or raise the cap.{% endif %}

+{% endif %} + +{% if target.reach != 'pending' %} +
+ {{ stat_tile("Series stored", target.kept) }} + {{ stat_tile("Over the cap", target.capped, state="degraded" if target.capped else none) }} + {{ stat_tile("Left out by filters", target.dropped) }} + {{ stat_tile("Scrape time", (target.scrape_ms|round|int ~ " ms") if target.scrape_ms is not none else "—") }} +
+{% endif %} +

Keeping {{ target.keep|join(", ") if target.keep else "everything" }}{% if target.drop %}, leaving out {{ target.drop|join(", ") }}{% endif %}. + Alert rules can watch any of these under Prometheus targets.

+ +
+
+

Series {{ total }}

+
+ + + {% if q %}Clear{% endif %} +
+
+ {% if not groups %} +

{% if q %}No stored metric matches “{{ q }}”.{% elif target.reach == 'pending' %}Nothing stored yet.{% else %}Nothing stored: the filters leave every series out, or the target hasn't answered.{% endif %}

+ {% else %} + + + + {% for g in groups %} + {% for r in g.rows %} + + {% if loop.first %}{% endif %} + + + + {% endfor %} + {% endfor %} + +
MetricLabelsNow
{{ g.metric }}{{ r.labels or "—" }} + {% if r.moment %}{{ r.moment|ago }}{% else %}{{ r.text }}{% endif %} +
+ {% endif %} +
+{% endblock %} diff --git a/plugins/snmp/service.py b/plugins/snmp/service.py index bb91d7b..634fc4e 100644 --- a/plugins/snmp/service.py +++ b/plugins/snmp/service.py @@ -6,6 +6,7 @@ from __future__ import annotations from dataclasses import dataclass from datetime import datetime +from steward.core.counter_rates import CounterRates from steward.core.crypto import decrypt_secret, encrypt_secret from .models import SnmpDevice @@ -168,28 +169,11 @@ def device_state(device: SnmpDevice, poll_seconds: int, now: datetime) -> Device return DeviceState("up", "Answering.") -class RateKeeper: - """A counter's previous sample per (device, key), to turn the next one - into a per-second rate. In memory: after a restart the first poll of each - counter stores nothing and the second stores its rate. A sample lower - than the one before (the device restarted, or a 32-bit counter wrapped) - is skipped rather than read as a huge negative or wrapped rate.""" - - def __init__(self) -> None: - self._last: dict[tuple[str, str], tuple[float, datetime]] = {} +class RateKeeper(CounterRates): + """A counter reading's previous sample per (device, key), its rate scaled + to the reading's unit (steward.core.counter_rates has the rules).""" def rate(self, device_id: str, reading: Reading, value: float, at: datetime) -> float | None: from .readings import COUNTER_UNITS - key = (device_id, reading.key) - prev = self._last.get(key) - self._last[key] = (value, at) - if prev is None: - return None - seconds = (at - prev[1]).total_seconds() - if seconds <= 0 or value < prev[0]: - return None - return (value - prev[0]) / seconds * COUNTER_UNITS[reading.unit][1] - - def forget(self, device_id: str) -> None: - for key in [k for k in self._last if k[0] == device_id]: - del self._last[key] + per_sec = super().rate((device_id, reading.key), value, at) + return None if per_sec is None else per_sec * COUNTER_UNITS[reading.unit][1] diff --git a/steward/core/counter_rates.py b/steward/core/counter_rates.py new file mode 100644 index 0000000..d7076b6 --- /dev/null +++ b/steward/core/counter_rates.py @@ -0,0 +1,34 @@ +"""A counter's previous sample, to turn the next one into a per-second rate. + +Scrapers that read monotonic counters (SNMP octets, Prometheus *_total) +store the rate, not the running total. Keys are tuples whose first element +is the owner (a device, a target), so forgetting an owner drops all of its +counters at once. +""" +from __future__ import annotations + +from datetime import datetime + + +class CounterRates: + """In memory: after a restart the first sample of each counter gives no + rate and the second does. A sample lower than the one before (the source + restarted, or a 32-bit counter wrapped) gives no rate rather than a huge + negative or wrapped one.""" + + def __init__(self) -> None: + self._last: dict[tuple, tuple[float, datetime]] = {} + + def rate(self, key: tuple, value: float, at: datetime) -> float | None: + prev = self._last.get(key) + self._last[key] = (value, at) + if prev is None: + return None + seconds = (at - prev[1]).total_seconds() + if seconds <= 0 or value < prev[0]: + return None + return (value - prev[0]) / seconds + + def forget(self, owner) -> None: + for key in [k for k in self._last if k[0] == owner]: + del self._last[key] diff --git a/steward/core/formatting.py b/steward/core/formatting.py index 1e54f6a..d1b81fe 100644 --- a/steward/core/formatting.py +++ b/steward/core/formatting.py @@ -245,7 +245,8 @@ def worst_reading(readings, thresholds: dict): def metric_value(name: str, value: Any) -> str: """A metric reading in its unit, from the name's suffix: _pct → 94%, _ms → 120 ms, _bytes → 1.2 GiB, _bps → its rate, _secs / seconds_ → a - span, _c → 71°C, _days → 12 days; is_up / is_healthy read as yes or no. + span (Prometheus's _seconds too), _per_sec → 1.5/s, _c → 71°C, _days → + 12 days; is_up / is_healthy read as yes or no. None (no reading) is "—".""" if value is None: return "—" @@ -260,7 +261,10 @@ def metric_value(name: str, value: Any) -> str: return human_bytes(value) if n.endswith(("_bps", "bytes_rate")): return human_rate(value) - if n.endswith("_secs") or n.startswith("seconds_"): + if n.endswith("_per_sec"): + return f"{value:.3g}/s" + if n.endswith("_secs") or n.startswith("seconds_") or ( + n.endswith("_seconds") and not n.endswith("_timestamp_seconds")): return human_duration(value) if n.endswith("_c"): return f"{value:.0f}°C" @@ -284,6 +288,9 @@ METRICS: dict[str, tuple[str, str]] = { "cleanup_secs": ("Hourly cleanup took", "how long the last retention and rollup run took"), "open_fds": ("Open files", "files Steward itself has open"), "open_fds_pct": ("Open files (% of limit)", "files Steward has open, against its limit"), + "scrape_ms": ("Scrape time", "how long the target took to hand over its metrics"), + "series_kept": ("Series stored", "series from the target's last scrape that were stored"), + "series_capped": ("Series over the cap", "series the last scrape left out: past the target's series cap"), "request_rate": ("Requests per second", "requests through this router"), "error_rate_4xx_pct": ("4xx responses", "share of responses that were client errors"), "error_rate_5xx_pct": ("5xx responses", "share of responses that were server errors"), @@ -321,7 +328,7 @@ SOURCES: dict[str, str] = { "icmp": "Ping monitors", "tcp": "TCP monitors", "dns": "DNS monitors", "http": "HTTP monitors", "steward": "Steward itself", "host_agent": "Host agent", "docker": "Docker containers", "traefik": "Traefik routers", "unifi": "UniFi", - "snmp": "SNMP devices", + "snmp": "SNMP devices", "prometheus": "Prometheus targets", } diff --git a/steward/core/settings.py b/steward/core/settings.py index 97e3879..3d4844d 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -142,6 +142,7 @@ DEFAULTS: dict[str, Any] = { "plugin.docker": {"enabled": True}, "plugin.host_agent": {"enabled": True}, "plugin.snmp": {"enabled": True}, + "plugin.prometheus": {"enabled": True}, # OIDC single-sign-on "oidc.enabled": False, "oidc.discovery_url": "", diff --git a/tests/core/test_metric_catalog.py b/tests/core/test_metric_catalog.py index a618847..a010134 100644 --- a/tests/core/test_metric_catalog.py +++ b/tests/core/test_metric_catalog.py @@ -9,7 +9,7 @@ import steward from steward.alerts.service import CORE_METRICS _REPO = pathlib.Path(steward.__file__).parent.parent -_PLUGINS = ("traefik", "unifi", "docker", "host_agent", "snmp") +_PLUGINS = ("traefik", "unifi", "docker", "host_agent", "snmp", "prometheus") def catalog() -> dict[str, set[str]]: @@ -98,5 +98,7 @@ def test_every_recorded_metric_is_in_the_catalog(): assert ("traefik", "error_rate_5xx_pct") in pairs and ("docker", "is_healthy") in pairs assert ("host_agent", "seconds_since_seen") in pairs cat = catalog() - missing = sorted(p for p in pairs if p[1] not in cat.get(p[0], set())) + # An empty list is a dynamic source (snmp, prometheus): the rule form + # offers whatever it has recorded, so any name it writes is offerable. + missing = sorted(p for p in pairs if cat.get(p[0]) != set() and p[1] not in cat.get(p[0], set())) assert not missing, f"recorded but not in the alert catalog: {missing}" diff --git a/tests/integration/test_prometheus_scrape.py b/tests/integration/test_prometheus_scrape.py new file mode 100644 index 0000000..8bcc216 --- /dev/null +++ b/tests/integration/test_prometheus_scrape.py @@ -0,0 +1,130 @@ +"""Integration: a Prometheus scrape writes each target's outcome to its row, +its kept series to the metric tables (a counter as a rate), and its own +is_up, which an alert rule can watch (#5668). Requires STEWARD_DATABASE_URL.""" +from __future__ import annotations + +import asyncio +import importlib +import os +import uuid + +import pytest + +pytestmark = pytest.mark.integration + +TEXT = """\ +# TYPE fa_backup_target_state gauge +fa_backup_target_state{target="Steward",state="late"} 0 +# TYPE fa_requests_total counter +fa_requests_total {count} +# TYPE noise gauge +noise 1 +""" + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +def _prom(sub): + """A submodule of the prometheus plugin as the app loaded it: it is on by + default, so its models are already mapped under the loader's name, and + importing plugins.prometheus.models would define the table a second time.""" + from steward.core.plugin_manager import get_loaded_plugin_modules + loaded = get_loaded_plugin_modules().get("prometheus") + return importlib.import_module(f"{loaded.__name__ if loaded else 'plugins.prometheus'}.{sub}") + + +def test_a_scrape_records_series_rates_and_a_down_target_fires_its_rule(app): + from sqlalchemy import select + + from steward.alerts import service as alert_svc + from steward.core import prometheus + from steward.core.alerts import record_metrics + from steward.core.crypto import encrypt_secret + from steward.models.alerts import AlertState, AlertStateEnum + from steward.models.metrics import PluginMetric, PluginMetricLatest + from steward.models.users import User, UserRole + + scheduler = _prom("scheduler") + PrometheusTarget = _prom("models").PrometheusTarget + ScrapeError = _prom("scraper").ScrapeError + + tag = uuid.uuid4().hex[:8] + up, down = f"arch-{tag}", f"gone-{tag}" + counts = iter([1_000, 1_011]) + tokens = [] + + async def fake_scrape(url, token, verify_tls): + tokens.append(token) + if "gone" in url: + raise ScrapeError("Couldn't connect: connection refused.") + return prometheus.parse(TEXT.format(count=next(counts))) + + async def latest(name_prefix): + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(PluginMetricLatest).where( + PluginMetricLatest.source_module == "prometheus", + PluginMetricLatest.resource_name.startswith(name_prefix)))).scalars() + return {(r.resource_name, r.metric_name): r.value for r in rows} + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"pr-{tag}", email=f"pr-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add_all([user, + PrometheusTarget(name=up, url=f"http://{up}/metrics", + token=encrypt_secret("scrape-token"), keep=["fa_*"]), + PrometheusTarget(name=down, url=f"http://{down}/metrics")]) + await s.flush() + await record_metrics(s, [PluginMetric(source_module="prometheus", resource_name=down, + metric_name="is_up", value=1.0)]) + await s.flush() + fields = await alert_svc.rule_fields( + s, name=f"down {tag}", source_module="prometheus", resource_name=down, + metric_name="is_up", operator="<", threshold=1) + rule = await alert_svc.add_rule(s, fields, created_by=user.id) + rule_id = rule.id + + await scheduler.scrape_due(app, scrape=fake_scrape) + first = await latest(up) + # Every target was just scraped, so nothing is due on the next tick. + await scheduler.scrape_due(app, scrape=fake_scrape) + assert len(tokens) == 2 + async with app.db_sessionmaker() as s: + async with s.begin(): + for t in (await s.execute(select(PrometheusTarget).where( + PrometheusTarget.name.in_([up, down])))).scalars(): + t.last_scraped_at = t.last_scraped_at.replace(year=2000) + await asyncio.sleep(1.1) + await scheduler.scrape_due(app, scrape=fake_scrape) + second = await latest(up) + async with app.db_sessionmaker() as s: + rows = {t.name: t for t in (await s.execute(select(PrometheusTarget).where( + PrometheusTarget.name.in_([up, down])))).scalars()} + state = (await s.execute(select(AlertState).where( + AlertState.rule_id == rule_id))).scalar_one() + async with app.db_sessionmaker() as s: + async with s.begin(): + for t in rows.values(): + await s.delete(await s.get(PrometheusTarget, t.id)) + return first, second, rows, state, await latest(down) + + first, second, rows, state, down_latest = asyncio.run(go()) + assert "scrape-token" in tokens + assert first[(f"{up}:target=Steward,state=late", "fa_backup_target_state")] == 0.0 + assert (up, "fa_requests_per_sec") not in first # a counter needs two scrapes + assert first[(up, "is_up")] == 1.0 and first[(up, "series_kept")] == 2.0 + assert not any(m == "noise" for _, m in first) # not in the keep list + # 11 requests over the 1.1 s or more between the scrapes. + assert 0 < second[(up, "fa_requests_per_sec")] <= 10.0 + assert rows[up].last_error is None and rows[up].last_kept == 2 and rows[up].last_dropped == 1 + assert rows[down].last_error == "Couldn't connect: connection refused." + assert rows[down].last_ok_at is None + assert down_latest[(down, "is_up")] == 0.0 + assert state.state == AlertStateEnum.firing diff --git a/tests/plugins/prometheus/__init__.py b/tests/plugins/prometheus/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugins/prometheus/test_pages.py b/tests/plugins/prometheus/test_pages.py new file mode 100644 index 0000000..366dd84 --- /dev/null +++ b/tests/plugins/prometheus/test_pages.py @@ -0,0 +1,100 @@ +"""The Prometheus pages (#5669): targets not answering first and saying why, +what was left out, the series a target stored, and the editor.""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +from plugins.prometheus.routes import group_series, shown_value +from tests._templates import render_plugin_template +from tests.test_style_tokens import hex_colours + + +def _view(name, reach, says="Answering.", ok_minutes=None, capped=0): + now = datetime.now(timezone.utc) + return {"id": f"id-{name}", "name": name, "url": f"http://{name}:8081/metrics", "reach": reach, + "says": says, "scraped_at": now - timedelta(seconds=20), + "ok_at": now - timedelta(minutes=ok_minutes) if ok_minutes is not None else None, + "interval": 60, "kept": 120, "dropped": 30, "capped": capped, "cap": 1000, + "keep": ["fabledarchives_*"], "drop": [], "scrape_ms": 42.4} + + +async def test_a_target_not_answering_says_why_and_the_cap_is_shown(app): + from quart import session + targets = [_view("archives", "down", "Answered 401: the token was refused.", ok_minutes=180), + _view("busy", "up", capped=37), _view("new", "pending")] + async with app.test_request_context("/plugins/prometheus/"): + session["user_role"] = "admin" + page = await render_plugin_template(app, "prometheus", "prometheus/index.html", targets=targets) + page = " ".join(page.split()) + assert 'Not answering' in page + assert "Answered 401: the token was refused." in page and "Last answered" in page + assert "37 series weren't stored" in page and "the cap of 1000" in page + assert '42 ms' in page + assert "the first scrape runs within seconds of saving" in page + assert 'href="/plugins/prometheus/targets/new"' in page and 'href="/plugins/prometheus/targets/id-busy/edit"' in page + main = page[page.index("Prometheus' in page # its breadcrumb + assert "/edit" not in page + + +async def test_the_form_never_echoes_the_token_and_offers_test(app): + from quart import session + target = NS(id="t1", name="archives") + v = {"name": "archives", "url": "http://archives:8081/metrics", "interval_seconds": 60, + "series_cap": 1000, "keep": "fabledarchives_*", "drop": "", "verify_tls": True, + "enabled": True, "has_token": True} + async with app.test_request_context("/plugins/prometheus/targets/t1/edit"): + session["user_role"] = "admin" + page = await render_plugin_template(app, "prometheus", "prometheus/form.html", + target=target, v=v, error=None) + assert 'name="token" type="password"' in page and "kept; type to change" in page + assert 'name="clear_token"' in page and 'hx-post="/plugins/prometheus/targets/test"' in page + assert "fabledarchives_*" in page and "Delete target" in page diff --git a/tests/plugins/prometheus/test_service.py b/tests/plugins/prometheus/test_service.py new file mode 100644 index 0000000..d5d1f54 --- /dev/null +++ b/tests/plugins/prometheus/test_service.py @@ -0,0 +1,150 @@ +"""The Prometheus plugin's pure parts: which series are stored and under what +names, the form, and what a page says about the newest scrape (#5668).""" +from datetime import datetime, timedelta, timezone + +import httpx +import pytest + +from plugins.prometheus.models import PrometheusTarget +from plugins.prometheus.scheduler import is_due +from plugins.prometheus.scraper import ScrapeError, scrape +from plugins.prometheus.service import (FormError, map_families, parse_form, patterns, + rate_name, resource_for, target_state, wanted) +from steward.core import prometheus +from steward.core.counter_rates import CounterRates + +TEXT = """\ +# TYPE fa_backup_target_state gauge +fa_backup_target_state{target="Steward",state="fresh"} 1 +fa_backup_target_state{target="Steward",state="late"} 0 +# TYPE fa_http_requests_total counter +fa_http_requests_total{route="GET /",status="200"} 100 +# TYPE fa_job_seconds histogram +fa_job_seconds_bucket{le="1"} 2 +fa_job_seconds_bucket{le="+Inf"} 3 +fa_job_seconds_sum 4.5 +fa_job_seconds_count 3 +# TYPE fa_rpc summary +fa_rpc{quantile="0.5"} 0.2 +fa_rpc_count 9 +fa_disk_free_bytes 1e9 +fa_broken NaN +other_thing 7 +""" + + +def _mapped(keep=(), drop=(), cap=1000, target="Archives"): + return map_families(target, prometheus.parse(TEXT), list(keep), list(drop), cap) + + +def test_gauges_stay_counters_become_rates_and_distributions_keep_sum_and_count(): + got = {(s.resource, s.metric): (s.value, s.counter) for s in _mapped().series} + assert got == { + ("Archives:target=Steward,state=fresh", "fa_backup_target_state"): (1.0, False), + ("Archives:target=Steward,state=late", "fa_backup_target_state"): (0.0, False), + ("Archives:route=GET /,status=200", "fa_http_requests_per_sec"): (100.0, True), + ("Archives", "fa_job_seconds_sum_per_sec"): (4.5, True), + ("Archives", "fa_job_seconds_count_per_sec"): (3.0, True), + ("Archives", "fa_rpc_count_per_sec"): (9.0, True), + ("Archives", "fa_disk_free_bytes"): (1e9, False), + ("Archives", "other_thing"): (7.0, False), + } + + +def test_keep_and_drop_patterns_choose_families_and_drop_wins(): + m = _mapped(keep=["fa_*"], drop=["fa_http_*"]) + assert {s.metric for s in m.series} == { + "fa_backup_target_state", "fa_job_seconds_sum_per_sec", "fa_job_seconds_count_per_sec", + "fa_rpc_count_per_sec", "fa_disk_free_bytes"} + assert m.dropped == 2 # the counter's one sample, other_thing's one + assert wanted("x", [], []) and not wanted("x", ["y*"], []) and not wanted("y1", ["y*"], ["y1"]) + + +def test_the_series_cap_and_overlong_names_are_counted_not_stored(): + m = _mapped(cap=3) + assert len(m.series) == 3 and m.capped == 5 + long = map_families("t" * 250, prometheus.parse('m{a="bcdefgh"} 1\n'), [], [], 10) + assert long.series == [] and long.capped == 1 + + +def test_names(): + assert rate_name("jobs_total") == "jobs_per_sec" + assert rate_name("job_seconds_sum") == "job_seconds_sum_per_sec" + assert resource_for("T", ()) == "T" + assert resource_for("T", (("a", "1"), ("b", "x"))) == "T:a=1,b=x" + assert patterns("a_*, b_*\n\n a_* \nc") == ["a_*", "b_*", "c"] + + +def test_a_counter_becomes_a_rate_and_a_reset_is_skipped(): + rates = CounterRates() + t0 = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc) + key = ("t1", "Archives", "x_per_sec") + assert rates.rate(key, 100, t0) is None + assert rates.rate(key, 160, t0 + timedelta(seconds=60)) == 1.0 + assert rates.rate(key, 5, t0 + timedelta(seconds=120)) is None # the service restarted + assert rates.rate(key, 35, t0 + timedelta(seconds=150)) == 1.0 + rates.forget("t1") + assert rates.rate(key, 99, t0 + timedelta(seconds=180)) is None + + +class _Form(dict): + def getlist(self, key): + v = self.get(key) + return v if isinstance(v, list) else ([v] if v else []) + + +def test_the_form_checks_what_it_stores_and_keeps_a_blank_token(): + values = parse_form(_Form(name="Archives", url="http://archives:8081/metrics", token="s3cret", + interval_seconds="30", series_cap="500", keep="fabledarchives_*", + enabled="on", verify_tls="on")) + assert values["token"].startswith("enc:") or values["token"] == "s3cret" + assert (values["interval_seconds"], values["series_cap"], values["keep"], values["enabled"]) == ( + 30, 500, ["fabledarchives_*"], True) + existing = PrometheusTarget(name="Archives", url="http://a/metrics", token="enc:v1:kept") + again = parse_form(_Form(name="Archives", url="http://a/metrics"), existing) + assert again["token"] == "enc:v1:kept" and again["enabled"] is False + cleared = parse_form(_Form(name="Archives", url="http://a/metrics", clear_token="on"), existing) + assert cleared["token"] == "" + for bad in (dict(name="", url="http://a"), dict(name="a:b", url="http://a"), + dict(name="a", url="ftp://a"), dict(name="a", url="http://a", interval_seconds="5"), + dict(name="a", url="http://a", series_cap="0")): + with pytest.raises(FormError): + parse_form(_Form(bad)) + + +def test_state_and_due(): + now = datetime(2026, 10, 12, 12, 0, tzinfo=timezone.utc) + t = PrometheusTarget(name="a", url="http://a", enabled=True, interval_seconds=60) + assert target_state(t, now).reach == "pending" and is_due(t, now) + t.last_scraped_at = now - timedelta(seconds=30) + assert target_state(t, now).reach == "up" and not is_due(t, now) + t.last_scraped_at = now - timedelta(seconds=58) + assert is_due(t, now) # half a tick of slack + t.last_error = "Couldn't connect: refused." + t.last_scraped_at = now - timedelta(seconds=30) + assert (target_state(t, now).reach, target_state(t, now).says) == ("down", "Couldn't connect: refused.") + t.last_scraped_at = now - timedelta(minutes=10) + assert target_state(t, now).reach == "stale" + t.enabled = False + assert target_state(t, now).reach == "off" and not is_due(t, now) + + +@pytest.mark.asyncio +async def test_a_scrape_sends_the_token_and_says_why_it_failed(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["auth"] = request.headers.get("Authorization") + if request.url.path == "/denied": + return httpx.Response(401) + if request.url.path == "/html": + return httpx.Response(200, text="hello") + return httpx.Response(200, text="up_thing 1\n") + + transport = httpx.MockTransport(handler) + fams = await scrape("http://svc/metrics", "tok", transport=transport) + assert list(fams) == ["up_thing"] and seen["auth"] == "Bearer tok" + with pytest.raises(ScrapeError, match="token was refused"): + await scrape("http://svc/denied", "bad", transport=transport) + with pytest.raises(ScrapeError, match="no metrics"): + await scrape("http://svc/html", transport=transport)