diff --git a/plugins/syslog/__init__.py b/plugins/syslog/__init__.py index 50029ff..23003ae 100644 --- a/plugins/syslog/__init__.py +++ b/plugins/syslog/__init__.py @@ -11,7 +11,7 @@ _app: "Quart | None" = None def setup(app: "Quart") -> None: global _app _app = app - from .models import SyslogLine, SyslogSource # noqa: F401 — registers the tables + from .models import SyslogLine, SyslogPattern, SyslogSource # noqa: F401 — registers the tables from .receiver import close # The listener is bound by the receive task, so it follows the settings @@ -20,9 +20,10 @@ def setup(app: "Quart") -> None: def get_scheduled_tasks() -> list: + from .patterns import make_count_task from .receiver import make_receive_task from .retention import make_prune_task - return [make_receive_task(_app), make_prune_task(_app)] + return [make_receive_task(_app), make_prune_task(_app), make_count_task(_app)] def get_blueprint(): @@ -39,7 +40,14 @@ def register_mcp_tools(register) -> None: register_tools(register) +def get_alert_metrics() -> dict[str, list[str]]: + """A pattern's count is named for the pattern (lines_matching_), so + the list is empty here and the rule form offers the counts recorded so far.""" + return {"syslog": []} + + def get_audit_entities() -> dict: - """Audit rows about a source name it and link to the Syslog page.""" - from .models import SyslogSource - return {"syslog_source": (SyslogSource, "/plugins/syslog/sources/{}")} + """Audit rows about a source or a pattern name it and link to its page.""" + from .models import SyslogPattern, SyslogSource + return {"syslog_source": (SyslogSource, "/plugins/syslog/sources/{}"), + "syslog_pattern": (SyslogPattern, "/plugins/syslog/patterns/{}/edit")} diff --git a/plugins/syslog/migrations/versions/syslog_002_patterns.py b/plugins/syslog/migrations/versions/syslog_002_patterns.py new file mode 100644 index 0000000..6d92618 --- /dev/null +++ b/plugins/syslog/migrations/versions/syslog_002_patterns.py @@ -0,0 +1,35 @@ +"""Syslog: telling phrases counted per source into the metric store (#6001) + +Revision ID: syslog_002_patterns +Revises: syslog_001_sources_lines +Create Date: 2026-10-19 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "syslog_002_patterns" +down_revision: Union[str, None] = "syslog_001_sources_lines" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "syslog_patterns", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("key", sa.String(48), nullable=False, unique=True), + sa.Column("name", sa.String(100), nullable=False), + sa.Column("phrases", sa.JSON, nullable=False), + sa.Column("max_severity", sa.SmallInteger, nullable=True), + sa.Column("source_id", sa.String(36), + sa.ForeignKey("syslog_sources.id", ondelete="CASCADE"), nullable=True), + sa.Column("window_minutes", sa.SmallInteger, nullable=False, server_default="15"), + sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.true()), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("syslog_patterns") diff --git a/plugins/syslog/models.py b/plugins/syslog/models.py index d5723ca..4b6a3ee 100644 --- a/plugins/syslog/models.py +++ b/plugins/syslog/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime, timezone -from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, SmallInteger, String, Text +from sqlalchemy import JSON, BigInteger, Boolean, DateTime, ForeignKey, Index, SmallInteger, String, Text from sqlalchemy.orm import Mapped, mapped_column from steward.core.ids import time_ordered_id @@ -65,3 +65,33 @@ class SyslogLine(Base): STATES = ("offered", "accepted", "ignored") + + +class SyslogPattern(Base): + """A telling phrase to count (#6001): lines whose message holds any of + `phrases` (case-insensitive) and, when `max_severity` is set, whose + severity is that or worse. Every minute each added source it covers (all, + or `source_id`'s) gets a sample of how many of its lines matched in the + last `window_minutes`, as metric `lines_matching_`, so an ordinary + alert rule fires on a match and clears once the window passes it.""" + __tablename__ = "syslog_patterns" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + key: Mapped[str] = mapped_column(String(48), nullable=False, unique=True) + name: Mapped[str] = mapped_column(String(100), nullable=False) + phrases: Mapped[list] = mapped_column(JSON, nullable=False, default=list) + max_severity: Mapped[int | None] = mapped_column(SmallInteger, nullable=True) + source_id: Mapped[str | None] = mapped_column( + String(36), ForeignKey("syslog_sources.id", ondelete="CASCADE"), nullable=True) + window_minutes: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=15) + 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)) + + @property + def metric(self) -> str: + return metric_for(self.key) + + +def metric_for(key: str) -> str: + return f"lines_matching_{key}" diff --git a/plugins/syslog/patterns.py b/plugins/syslog/patterns.py new file mode 100644 index 0000000..3509818 --- /dev/null +++ b/plugins/syslog/patterns.py @@ -0,0 +1,262 @@ +# plugins/syslog/patterns.py +"""Telling phrases in the kept lines, counted per source into the metric +store every minute (#6001), so an ordinary alert rule can watch them. + +A pattern's sample is the number of a source's lines that matched in the last +`window_minutes`, written for every added source it covers, zero included: +a rule "> 0" fires on the first match and clears once the window has passed +it, rather than holding the last count forever. + +Phrases are plain text matched case-insensitively anywhere in the message, +not regular expressions: they read the same in the form, the page and the +query, and a phrase someone types can't make the count slow. +""" +from __future__ import annotations + +import asyncio +import logging +import re +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING + +from sqlalchemy import func, or_, select + +from steward.core.like import escape_like +from steward.core.scheduler import ScheduledTask + +from .models import SyslogLine, SyslogPattern, SyslogSource, metric_for + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + +SOURCE_MODULE = "syslog" +COUNT_S = 60 +# The count's own deadline: past it the minute is skipped, not stacked. +COUNT_DEADLINE_S = 45.0 +WINDOWS = (5, 15, 60, 240, 1440) +MAX_PHRASES = 20 + + +@dataclass(frozen=True) +class Preset: + key: str + name: str + phrases: tuple[str, ...] = () + max_severity: int | None = None + about: str = "" + + +# Storage trouble as NAS and Linux kernels word it, and errors of any kind. +PRESETS: tuple[Preset, ...] = ( + Preset("io_error", "Disk I/O errors", + ("I/O error", "blk_update_request", "medium error", "end_request:"), + about="A disk or its controller failed a read or write."), + Preset("disk_reset", "Disk resets and timeouts", + ("hard resetting link", "SATA link down", "exception Emask", "link is slow to respond", + "device reset"), + about="A disk stopped answering and the kernel reset its link."), + Preset("nfs", "NFS server trouble", + ("nfsd:", "nfs: server", "not responding, still trying", "lockd:"), + about="The NFS server or its lock manager complained."), + Preset("fs_error", "Filesystem errors", + ("EXT4-fs error", "BTRFS error", "BTRFS critical", "XFS: ", "Remounting filesystem read-only", + "csum failed"), + about="A filesystem found damage or went read-only."), + Preset("raid", "RAID or storage pool degraded", + ("degraded", "Disk failure on", "crashed", "faulty"), + about="An array lost a disk, or a storage pool crashed."), + Preset("errors", "Errors and worse", max_severity=3, + about="Any line the device marked err, crit, alert or emerg."), +) +PRESETS_BY_KEY = {p.key: p for p in PRESETS} + + +class PatternError(ValueError): + pass + + +_KEY = re.compile(r"[^a-z0-9]+") + + +def key_for(name: str) -> str: + """A pattern's metric key from its name: "Disk I/O errors" → "disk_i_o_errors".""" + return _KEY.sub("_", name.lower()).strip("_")[:48] or "pattern" + + +def parse_form(form) -> dict: + """A pattern's fields from its form, or PatternError saying what to fix.""" + name = (form.get("name") or "").strip()[:100] + if not name: + raise PatternError("Give the pattern a name.") + phrases = [p.strip() for p in (form.get("phrases") or "").splitlines() if p.strip()] + if len(phrases) > MAX_PHRASES: + raise PatternError(f"At most {MAX_PHRASES} phrases.") + if any(len(p) > 200 for p in phrases): + raise PatternError("A phrase can be at most 200 characters.") + severity = (form.get("max_severity") or "").strip() + max_severity = int(severity) if severity.isdigit() and int(severity) <= 7 else None + if not phrases and max_severity is None: + raise PatternError("Give at least one phrase, or a severity to count.") + try: + window = int(form.get("window_minutes") or 15) + except ValueError: + window = 15 + if window not in WINDOWS: + window = 15 + return {"name": name, "phrases": phrases, "max_severity": max_severity, + "source_id": form.get("source_id") or None, "window_minutes": window, + "enabled": "enabled" in form} + + +def matches(pattern, severity: int, message: str) -> bool: + """Whether one line matches: the same test the count's query makes.""" + if pattern.max_severity is not None and severity > pattern.max_severity: + return False + text = message.lower() + return not pattern.phrases or any(p.lower() in text for p in pattern.phrases) + + +def resource_of(source: SyslogSource) -> str: + """A source's name in the metric store.""" + return source.name or source.hostname_seen or source.address + + +def _match_query(p, source_ids: list[str]): + """The lines a pattern matches among these sources: the same test as matches().""" + stmt = select(SyslogLine.id, SyslogLine.source_id, SyslogLine.received_at).where( + SyslogLine.source_id.in_(source_ids)) + if p.max_severity is not None: + stmt = stmt.where(SyslogLine.severity <= p.max_severity) + if p.phrases: + stmt = stmt.where(or_(*(SyslogLine.message.ilike(f"%{escape_like(ph)}%", escape="\\") + for ph in p.phrases))) + return stmt + + +def _minute(at: datetime) -> datetime: + return at.replace(second=0, microsecond=0) + + +@dataclass +class Tally: + """A pattern's matches per source per minute, kept between counts so each + count reads only the lines stored since the last one, not its whole window.""" + sig: tuple + minutes: dict[str, Counter] = field(default_factory=dict) + + def add(self, source_id: str, received_at: datetime) -> None: + self.minutes.setdefault(source_id, Counter())[_minute(received_at)] += 1 + + def count(self, source_id: str, since: datetime) -> int: + """Matches from `since`'s minute on; minutes before it are dropped.""" + per = self.minutes.get(source_id) + if not per: + return 0 + start = _minute(since) + for m in [m for m in per if m < start]: + del per[m] + return sum(per.values()) + + +@dataclass +class Tallies: + """Every pattern's tally, and the newest line id they have all counted. + Line ids are time-ordered and given as a line is stored, by the one + receiver, so lines stored later always have higher ids: a line held + through a database outage is still counted when it lands, in the minute + it was received.""" + upto: str | None = None + by_pattern: dict[str, Tally] = field(default_factory=dict) + + +TALLIES = Tallies() + + +def _sig(p) -> tuple: + return (tuple(p.phrases or ()), p.max_severity, p.source_id, p.window_minutes) + + +async def count_matches(session, patterns: list, sources: list, now: datetime, + tallies: Tallies | None = None) -> list[tuple[str, str, int]]: + """(resource, metric, lines matched in the pattern's window) for every + enabled pattern and every added source it covers, zero included. + + A pattern new or changed since the last count reads its window once; + after that, each count reads only the lines stored since the one before.""" + tallies = TALLIES if tallies is None else tallies + try: + return await _count(session, patterns, sources, now, tallies) + except BaseException: + # Cut short (the count's deadline, a lost connection): some tallies + # may hold lines the cursor doesn't yet cover, so start them over. + tallies.upto, tallies.by_pattern = None, {} + raise + + +async def _count(session, patterns: list, sources: list, now: datetime, tallies: Tallies) -> list: + newest = (await session.execute(select(func.max(SyslogLine.id)))).scalar() + out = [] + live = set() + for p in patterns: + if not p.enabled: + continue + live.add(p.id) + covered = [s for s in sources if s.state == "accepted" and p.source_id in (None, s.id)] + if not covered: + continue + since = now - timedelta(minutes=p.window_minutes) + tally = tallies.by_pattern.get(p.id) + stmt = _match_query(p, [s.id for s in covered]) + if tally is None or tally.sig != _sig(p) or tallies.upto is None: + tally = tallies.by_pattern[p.id] = Tally(_sig(p)) + stmt = stmt.where(SyslogLine.received_at >= since) + else: + stmt = stmt.where(SyslogLine.id > tallies.upto) + if newest is not None: + stmt = stmt.where(SyslogLine.id <= newest) + for _, source_id, received_at in (await session.execute(stmt)).all(): + tally.add(source_id, received_at) + out.extend((resource_of(s), metric_for(p.key), tally.count(s.id, since)) for s in covered) + for gone in set(tallies.by_pattern) - live: + del tallies.by_pattern[gone] + if newest is not None: + tallies.upto = newest + return out + + +def make_count_task(app: "Quart") -> ScheduledTask: + async def count() -> None: + try: + await asyncio.wait_for(record_counts(app), COUNT_DEADLINE_S) + except TimeoutError: + logger.warning("Syslog: counting pattern matches took over %.0f s; this minute is skipped", + COUNT_DEADLINE_S) + + return ScheduledTask(name="syslog_pattern_counts", coro_factory=count, interval_seconds=COUNT_S) + + +async def record_counts(app: "Quart", now: datetime | None = None) -> int: + """Count every pattern's matches and record them as metrics, so the alert + rules watching them are evaluated. Returns how many samples it wrote.""" + from steward.core.alerts import MetricBatch + + now = now or datetime.now(timezone.utc) + async with app.db_sessionmaker() as session: + async with session.begin(): + patterns = list((await session.execute( + select(SyslogPattern).where(SyslogPattern.enabled.is_(True)))).scalars()) + if not patterns: + return 0 + sources = list((await session.execute( + select(SyslogSource).where(SyslogSource.state == "accepted"))).scalars()) + samples = await count_matches(session, patterns, sources, now) + batch = MetricBatch() + for resource, metric, value in samples: + batch.add(source_module=SOURCE_MODULE, resource_name=resource, metric_name=metric, + value=float(value), recorded_at=now) + await batch.record(session) + return len(samples) diff --git a/plugins/syslog/routes.py b/plugins/syslog/routes.py index 7715849..74a3a85 100644 --- a/plugins/syslog/routes.py +++ b/plugins/syslog/routes.py @@ -9,15 +9,19 @@ from __future__ import annotations from datetime import datetime, timezone from quart import Blueprint, abort, current_app, redirect, render_template, request, session -from sqlalchemy import select +from sqlalchemy import delete, select from steward.auth.middleware import require_role from steward.core.audit import log_audit +from steward.core.metric_series import latest_by_resource from steward.core.time_range import RANGES +from steward.models.metrics import PluginMetricLatest from steward.models.users import UserRole -from .models import SyslogLine, SyslogSource -from .parse import facility_name, severity_name +from .models import SyslogLine, SyslogPattern, SyslogSource +from .parse import SEVERITIES, facility_name, severity_name +from .patterns import (PRESETS, PRESETS_BY_KEY, SOURCE_MODULE, WINDOWS, PatternError, key_for, parse_form, + resource_of) from .receiver import LISTENER, MAX_SENDERS, settings from .service import PANEL_LINES, PANEL_SPAN, SEVERITY_CHOICES, apps_of, filtered, line_filter, rates, sources_for @@ -93,6 +97,27 @@ async def _source_or_404(db, source_id: str) -> SyslogSource: return source +async def _unique_name(db, source: SyslogSource, wanted: str) -> str: + """The name a source is given, made its own: a source's pattern counts + and the alert rules on them are filed under its name, so two sources + sharing one would read as one. A name taken is kept with the address.""" + name = wanted.strip()[:100] or source.hostname_seen or source.address + taken = (await db.execute(select(SyslogSource.id).where( + SyslogSource.name == name, SyslogSource.id != source.id).limit(1))).first() + return f"{name[:80]} ({source.address})" if taken else name + + +async def _move_resource(db, old: str, new: str) -> None: + """Move a renamed source's counts, and the alert rules on them, to its new name.""" + from sqlalchemy import update + + from steward.models.alerts import AlertRule + from steward.models.metrics import PluginMetric, PluginMetricLatest + for model in (PluginMetric, PluginMetricLatest, AlertRule): + await db.execute(update(model).where( + model.source_module == SOURCE_MODULE, model.resource_name == old).values(resource_name=new)) + + async def _host_id(db, value: str | None) -> str | None: from steward.models.hosts import Host if not value: @@ -108,7 +133,7 @@ async def add_source(source_id: str): async with current_app.db_sessionmaker() as db: async with db.begin(): source = await _source_or_404(db, source_id) - source.name = (form.get("name") or "").strip()[:100] or source.hostname_seen or source.address + source.name = await _unique_name(db, source, form.get("name") or "") source.host_id = await _host_id(db, form.get("host_id")) source.state = "accepted" detail = {"name": source.name, "address": source.address} @@ -159,12 +184,15 @@ async def save_source(source_id: str): async with current_app.db_sessionmaker() as db: async with db.begin(): source = await _source_or_404(db, source_id) + old_resource = resource_of(source) old_name = source.name - source.name = (form.get("name") or "").strip()[:100] or source.hostname_seen or source.address + source.name = await _unique_name(db, source, form.get("name") or "") source.host_id = await _host_id(db, form.get("host_id")) detail = {"name": source.name} if old_name != source.name: detail["old_name"] = old_name + if old_resource != resource_of(source): + await _move_resource(db, old_resource, resource_of(source)) await log_audit(current_app, *_actor(), "syslog.source_updated", entity_type="syslog_source", entity_id=source_id, detail=detail) return redirect(f"/plugins/syslog/sources/{source_id}") @@ -179,7 +207,11 @@ async def delete_source(source_id: str): async with db.begin(): source = await _source_or_404(db, source_id) detail = {"name": source.name, "address": source.address} + resource = resource_of(source) await db.delete(source) + # Its newest pattern counts go with it; its history ages out. + await db.execute(delete(PluginMetricLatest).where( + PluginMetricLatest.source_module == SOURCE_MODULE, PluginMetricLatest.resource_name == resource)) await log_audit(current_app, *_actor(), "syslog.source_removed", entity_type="syslog_source", entity_id=source_id, detail=detail) return redirect(INDEX) @@ -261,3 +293,167 @@ async def lines_panel(): rows = [{**line_view(ln), "source": names[ln.source_id], "source_id": ln.source_id} for ln in lines] return await render_template("syslog/panel.html", lines=rows, sources=[{"id": sid, "name": name} for sid, name in names.items()]) + + +# ── Patterns: telling phrases counted for alert rules (#6001) ───────────────── + +PATTERNS = "/plugins/syslog/patterns" + + +def pattern_view(p: SyslogPattern, sources: dict, latest: dict) -> dict: + covered = [s for s in sources.values() if s.state == "accepted" and p.source_id in (None, s.id)] + return {"id": p.id, "key": p.key, "name": p.name, "metric": p.metric, "phrases": p.phrases or [], + "severity": severity_name(p.max_severity) if p.max_severity is not None else None, + "scope": sources[p.source_id].name or sources[p.source_id].address + if p.source_id in sources else None, + "window": p.window_minutes, "enabled": p.enabled, + "counts": [{"source": resource_of(s), "value": _value(latest, resource_of(s), p.metric)} for s in covered], + "alert_resource": resource_of(covered[0]) if covered else None} + + +def _value(latest: dict, resource: str, metric: str) -> int | None: + row = latest.get(resource, {}).get(metric) + return int(row.value) if row is not None else None + + +@syslog_bp.get("/patterns") +@require_role(UserRole.viewer) +async def patterns_index(): + async with current_app.db_sessionmaker() as db: + patterns = list((await db.execute(select(SyslogPattern).order_by(SyslogPattern.name))).scalars()) + sources = {s.id: s for s in (await db.execute(select(SyslogSource))).scalars()} + latest = await latest_by_resource(db, SOURCE_MODULE, {resource_of(s) for s in sources.values()}, + [p.metric for p in patterns]) + taken = {p.key for p in patterns} + return await render_template("syslog/patterns.html", patterns=[pattern_view(p, sources, latest) for p in patterns], + presets=[p for p in PRESETS if p.key not in taken], + any_sources=any(s.state == "accepted" for s in sources.values())) + + +async def _free_key(db, wanted: str) -> str: + taken = set((await db.execute(select(SyslogPattern.key))).scalars()) + key, n = wanted, 2 + while key in taken: + key = f"{wanted[:44]}_{n}" + n += 1 + return key + + +async def _accepted_sources(db) -> list: + return list((await db.execute(select(SyslogSource).where(SyslogSource.state == "accepted") + .order_by(SyslogSource.name))).scalars()) + + +async def _render_pattern_form(pattern, values: dict, error: str | None = None, status: int = 200): + async with current_app.db_sessionmaker() as db: + sources = await _accepted_sources(db) + return await render_template("syslog/pattern_form.html", pattern=pattern, v=values, error=error, + sources=sources, windows=WINDOWS, severities=list(enumerate(SEVERITIES))), status + + +@syslog_bp.post("/patterns/preset/") +@require_role(UserRole.admin) +async def add_preset(key: str): + preset = PRESETS_BY_KEY.get(key) + if preset is None: + abort(404) + async with current_app.db_sessionmaker() as db: + async with db.begin(): + pattern = SyslogPattern(key=await _free_key(db, preset.key), name=preset.name, + phrases=list(preset.phrases), max_severity=preset.max_severity) + db.add(pattern) + await db.flush() + pattern_id = pattern.id + await log_audit(current_app, *_actor(), "syslog.pattern_added", entity_type="syslog_pattern", + entity_id=pattern_id, detail={"name": preset.name, "preset": preset.key}) + return redirect(PATTERNS) + + +@syslog_bp.get("/patterns/new") +@require_role(UserRole.admin) +async def new_pattern(): + return await _render_pattern_form(None, {"window_minutes": 15, "enabled": True}) + + +def _typed_pattern(form) -> dict: + return {"name": form.get("name", ""), "phrases": form.get("phrases", ""), + "max_severity": form.get("max_severity", ""), "source_id": form.get("source_id", ""), + "window_minutes": form.get("window_minutes", 15, type=int), "enabled": "enabled" in form} + + +@syslog_bp.post("/patterns/new") +@require_role(UserRole.admin) +async def create_pattern(): + form = await request.form + try: + values = parse_form(form) + except PatternError as exc: + return await _render_pattern_form(None, _typed_pattern(form), str(exc), 422) + async with current_app.db_sessionmaker() as db: + async with db.begin(): + if values["source_id"] and await db.get(SyslogSource, values["source_id"]) is None: + values["source_id"] = None + pattern = SyslogPattern(key=await _free_key(db, key_for(values["name"])), **values) + db.add(pattern) + await db.flush() + pattern_id = pattern.id + await log_audit(current_app, *_actor(), "syslog.pattern_added", entity_type="syslog_pattern", + entity_id=pattern_id, detail={"name": values["name"]}) + return redirect(PATTERNS) + + +async def _pattern_or_404(db, pattern_id: str) -> SyslogPattern: + pattern = await db.get(SyslogPattern, pattern_id) + if pattern is None: + abort(404) + return pattern + + +@syslog_bp.get("/patterns//edit") +@require_role(UserRole.admin) +async def edit_pattern(pattern_id: str): + async with current_app.db_sessionmaker() as db: + p = await _pattern_or_404(db, pattern_id) + return await _render_pattern_form(p, { + "name": p.name, "phrases": "\n".join(p.phrases or []), + "max_severity": "" if p.max_severity is None else str(p.max_severity), + "source_id": p.source_id or "", "window_minutes": p.window_minutes, "enabled": p.enabled}) + + +@syslog_bp.post("/patterns//edit") +@require_role(UserRole.admin) +async def save_pattern(pattern_id: str): + """Change what a pattern matches. Its key, and so its metric and the + alert rules on it, stay as they were.""" + form = await request.form + async with current_app.db_sessionmaker() as db: + async with db.begin(): + p = await _pattern_or_404(db, pattern_id) + try: + values = parse_form(form) + except PatternError as exc: + return await _render_pattern_form(p, _typed_pattern(form), str(exc), 422) + if values["source_id"] and await db.get(SyslogSource, values["source_id"]) is None: + values["source_id"] = None + for column, value in values.items(): + setattr(p, column, value) + await log_audit(current_app, *_actor(), "syslog.pattern_updated", entity_type="syslog_pattern", + entity_id=pattern_id, detail={"name": values["name"]}) + return redirect(PATTERNS) + + +@syslog_bp.post("/patterns//delete") +@require_role(UserRole.admin) +async def delete_pattern(pattern_id: str): + """Remove a pattern. Its newest counts go with it, so nothing reads it as + a series that went quiet; its history ages out with retention.""" + async with current_app.db_sessionmaker() as db: + async with db.begin(): + p = await _pattern_or_404(db, pattern_id) + name, metric = p.name, p.metric + await db.delete(p) + await db.execute(delete(PluginMetricLatest).where( + PluginMetricLatest.source_module == SOURCE_MODULE, PluginMetricLatest.metric_name == metric)) + await log_audit(current_app, *_actor(), "syslog.pattern_removed", entity_type="syslog_pattern", + entity_id=pattern_id, detail={"name": name}) + return redirect(PATTERNS) diff --git a/plugins/syslog/templates/syslog/index.html b/plugins/syslog/templates/syslog/index.html index ecad2c4..6cddf1b 100644 --- a/plugins/syslog/templates/syslog/index.html +++ b/plugins/syslog/templates/syslog/index.html @@ -70,7 +70,10 @@ {% endif %}
-

Sources

+
+

Sources

+ Patterns +
{% if not sources %} {{ empty_state("No sources yet. Point a NAS, router or other device at this server's syslog port: it shows above as sending, and Add keeps its lines.") }} {% else %} diff --git a/plugins/syslog/templates/syslog/pattern_form.html b/plugins/syslog/templates/syslog/pattern_form.html new file mode 100644 index 0000000..938985d --- /dev/null +++ b/plugins/syslog/templates/syslog/pattern_form.html @@ -0,0 +1,72 @@ +{# plugins/syslog/templates/syslog/pattern_form.html — a pattern's phrases, + severity, source and window, and Remove (#6001). #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% set heading = ("Edit " ~ pattern.name) if pattern else "New pattern" %} +{% block title %}{{ heading }} — Syslog — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Syslog", "/plugins/syslog/"), ("Patterns", "/plugins/syslog/patterns"), (pattern.name if pattern else "New", "")]) }}{% endblock %} +{% block content %} +{{ page_header(heading) }} + +
+{% if error %}{% endif %} +
{{ csrf_field() }} +
+

Pattern

+
+
+ + + {% if pattern %}Alert rules watch {{ pattern.metric }}, which stays as it is when the name changes.{% endif %} +
+
+ + + One per line. A line matches if it contains any of them, in any case. Leave empty to count every line at the severity below. +
+
+ + +
+
+ + +
+
+ + + The count is of matches this far back, so an alert on it clears this long after the last one. +
+
+ +
+
+
+
+ + Cancel +
+
+ +{% if pattern %} +
+

Remove

+

Stops counting. An alert rule on it then reads no data.

+
{{ csrf_field() }} + +
+
+{% endif %} +
+{% endblock %} diff --git a/plugins/syslog/templates/syslog/patterns.html b/plugins/syslog/templates/syslog/patterns.html new file mode 100644 index 0000000..f61c6f3 --- /dev/null +++ b/plugins/syslog/templates/syslog/patterns.html @@ -0,0 +1,71 @@ +{# plugins/syslog/templates/syslog/patterns.html — telling phrases counted in + each source's lines every minute, so an alert rule can watch them (#6001). #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header, empty_state %} +{% block title %}Patterns — Syslog — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Syslog", "/plugins/syslog/"), ("Patterns", "")]) }}{% endblock %} +{% block content %} +{% set admin = session.user_role == 'admin' %} +{% set can_alert = session.user_role in ('admin', 'operator') %} +{{ page_header("Patterns", (patterns|length ~ " pattern" ~ ("s" if patterns|length != 1 else "")) if patterns else none) }} + +
+
+

Counted

+ {% if admin %}New pattern{% endif %} +
+

Each minute, every added source's lines matching a pattern in its window are counted. Alert on a count over 0 to hear of the first match; the alert clears once the window has passed it.

+ {% if not patterns %} + {{ empty_state("No patterns yet. Add one of the ready-made ones below, or write your own.") }} + {% else %} + + {% if can_alert %}{% endif %} + + {% for p in patterns %} + + + + + + {% if can_alert %} + + {% endif %} + + {% endfor %} + +
PatternMatchesWindowCount now
{{ p.name }}{% if not p.enabled %} (off){% endif %}
{{ p.metric }}
 + {% if p.phrases %}{{ p.phrases[:3]|join(" · ") }}{% if p.phrases|length > 3 %} · +{{ p.phrases|length - 3 }} more{% endif %}{% endif %} + {% if p.severity %}
{{ p.severity }} and worse
{% endif %} + {% if p.scope %}
from {{ p.scope }} only
{% endif %} + 		{{ p.window }} min + {% if not p.counts %}No added source + {% else %}{% for c in p.counts %}
{{ "—" if c.value is none else c.value }} {{ c.source }}
{% endfor %}{% endif %} + 		+ {% if p.alert_resource %}Alert on it{% endif %} + {% if admin %}Edit{% endif %} +
+ {% endif %} +
+ +{% if admin and presets %} +
+

Ready-made

+

How NAS and Linux kernels word storage trouble. Add one, then change its phrases if your devices word it differently.

+ + + {% for p in presets %} + + + + + + {% endfor %} + +
{{ p.name }}
{{ p.about }}
{{ p.phrases|join(" · ") if p.phrases else "err, crit, alert or emerg" }} +
{{ csrf_field() }} + +
+
+
+{% endif %} +{% endblock %} diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index cf73d44..9c52ea9 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -423,9 +423,14 @@ async def _cover_choices(db, source: str, rule: AlertRule | None = None) -> dict @alerts_bp.get("/rules/new") @require_role(UserRole.operator) async def new_rule(): - first_source = alert_svc.source_modules()[0] + """The rule form, empty, or aimed at a series when a page links here with + source_module, resource_name and metric_name (a syslog pattern's count).""" + sources = alert_svc.source_modules() + asked = request.args.get("source_module", "") + first_source = asked if asked in sources else sources[0] async with current_app.db_sessionmaker() as db: - ctx = {**await _cover_choices(db, first_source), **await _target(db, first_source)} + ctx = {**await _cover_choices(db, first_source), + **(await _target_from_args(db) if asked == first_source else await _target(db, first_source))} return await render_template( "alerts/rules_form.html", rule=None, **ctx, diff --git a/steward/audit/service.py b/steward/audit/service.py index 07005e4..537fabd 100644 --- a/steward/audit/service.py +++ b/steward/audit/service.py @@ -203,6 +203,9 @@ ACTIONS = { "syslog.source_removed": "removed syslog source", "syslog.source_ignored": "ignored syslog sender", "syslog.source_offered": "offered again syslog sender", + "syslog.pattern_added": "added syslog pattern", + "syslog.pattern_updated": "changed syslog pattern", + "syslog.pattern_removed": "removed syslog pattern", "backup.made": "sent a backup", "backup.failed": "tried to send a backup, which failed:", "backup.restored": "restored Steward from a backup", diff --git a/steward/backup/contents.py b/steward/backup/contents.py index 29f90fa..4ea7e97 100644 --- a/steward/backup/contents.py +++ b/steward/backup/contents.py @@ -28,6 +28,7 @@ KEPT: frozenset[str] = frozenset({ "unifi_known_clients", # Syslog senders: which were added, under what name, and which ignored. "syslog_sources", + "syslog_patterns", }) # Rows that are measurements, logs, history or in-flight work: left out. @@ -115,5 +116,5 @@ AREAS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("Ansible", ("ansible_targets", "ansible_groups", "ansible_target_groups", "ansible_schedules")), ("Dashboards", ("dashboards", "dashboard_widgets", "dashboard_share_tokens")), ("Plugins", ("snmp_devices", "prometheus_targets", "prometheus_widget_templates", - "unifi_known_clients", "syslog_sources")), + "unifi_known_clients", "syslog_sources", "syslog_patterns")), ) diff --git a/steward/core/formatting.py b/steward/core/formatting.py index 9b4dd39..26b4415 100644 --- a/steward/core/formatting.py +++ b/steward/core/formatting.py @@ -376,7 +376,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", "prometheus": "Prometheus targets", + "snmp": "SNMP devices", "prometheus": "Prometheus targets", "syslog": "Syslog patterns", } diff --git a/tests/integration/test_syslog.py b/tests/integration/test_syslog.py index 3935592..cd67cc8 100644 --- a/tests/integration/test_syslog.py +++ b/tests/integration/test_syslog.py @@ -228,3 +228,136 @@ def test_search_syslog_filters_and_pages(app): await s.execute(delete_sources()) asyncio.run(go()) + + +def _latest(app, metric): + async def read(): + from sqlalchemy import select + + from steward.models.metrics import PluginMetricLatest + async with app.db_sessionmaker() as s: + return {r.resource_name: r.value for r in (await s.execute(select(PluginMetricLatest).where( + PluginMetricLatest.source_module == "syslog", PluginMetricLatest.metric_name == metric))).scalars()} + return read() + + +async def _clear_patterns(app): + from sqlalchemy import delete + + from steward.models.metrics import PluginMetric, PluginMetricLatest + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(_mod("models").SyslogPattern)) + await s.execute(delete_sources()) + for model in (PluginMetric, PluginMetricLatest): + await s.execute(delete(model).where(model.source_module == "syslog")) + + +def test_patterns_count_matches_in_their_window_zeros_included(app): + from sqlalchemy import select + + models, patterns = _mod("models"), _mod("patterns") + now = datetime.now(timezone.utc) + + async def go(): + await _clear_patterns(app) + patterns.TALLIES = patterns.Tallies() + _, _, src_id, src_name = await _seed(app, now) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all([ + models.SyslogPattern(key="nfs", name="NFS", phrases=["NFSD:"], window_minutes=60), + models.SyslogPattern(key="errors", name="Errors", phrases=[], max_severity=3, window_minutes=15), + models.SyslogPattern(key="quota", name="Quota", phrases=["100%"], window_minutes=15), + models.SyslogPattern(key="off", name="Off", phrases=["eth0"], enabled=False)]) + assert await patterns.record_counts(app, now) == 3 + # Case-insensitive; the err 40 minutes ago is past a 15-minute window; + # % is literal, and a pattern counting nothing still writes its 0. + assert await _latest(app, "lines_matching_nfs") == {src_name: 1.0} + assert await _latest(app, "lines_matching_errors") == {src_name: 0.0} + assert await _latest(app, "lines_matching_quota") == {src_name: 0.0} + assert await _latest(app, "lines_matching_off") == {} + + # A line stored after the count is counted by the next, which reads + # only the new lines. + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(models.SyslogLine(source_id=src_id, received_at=now, facility=0, severity=2, + hostname="synnas", app="kernel", message="sd 0:0:0:0: I/O error")) + later = now + timedelta(minutes=1) + await patterns.record_counts(app, later) + assert await _latest(app, "lines_matching_errors") == {src_name: 1.0} + assert await _latest(app, "lines_matching_nfs") == {src_name: 1.0} # not counted twice + + # A changed pattern reads its window again. + async with app.db_sessionmaker() as s: + async with s.begin(): + p = (await s.execute(select(models.SyslogPattern).where(models.SyslogPattern.key == "errors"))).scalar_one() + p.window_minutes = 60 + await patterns.record_counts(app, later) + assert await _latest(app, "lines_matching_errors") == {src_name: 2.0} + # The window passes the matches and the count falls back to 0. + await patterns.record_counts(app, now + timedelta(hours=2)) + assert await _latest(app, "lines_matching_errors") == {src_name: 0.0} + await _clear_patterns(app) + + asyncio.run(go()) + + +def test_patterns_page_adds_edits_and_removes_and_a_rename_moves_counts(app): + from sqlalchemy import select + + models, patterns = _mod("models"), _mod("patterns") + csrf = {"X-CSRF-Token": "test-token"} + now = datetime.now(timezone.utc) + + async def go(): + await _clear_patterns(app) + patterns.TALLIES = patterns.Tallies() + _, _, src_id, src_name = await _seed(app, now) + admin = await _client(app, "admin") + assert (await admin.post("/plugins/syslog/patterns/preset/nfs", headers=csrf)).status_code == 302 + refused = await admin.post("/plugins/syslog/patterns/new", headers=csrf, + form={"name": "Nothing", "phrases": " \n", "window_minutes": "15"}) + assert refused.status_code == 422 + assert "at least one phrase" in await refused.get_data(as_text=True) + made = await admin.post("/plugins/syslog/patterns/new", headers=csrf, + form={"name": "Quota full", "phrases": "100%\nquota", "max_severity": "4", + "window_minutes": "60", "enabled": "on"}) + assert made.status_code == 302 + async with app.db_sessionmaker() as s: + rows = {p.key: p for p in (await s.execute(select(models.SyslogPattern))).scalars()} + assert set(rows) == {"nfs", "quota_full"} + assert (rows["quota_full"].phrases, rows["quota_full"].max_severity) == (["100%", "quota"], 4) + assert rows["nfs"].phrases[0] == "nfsd:" + + await patterns.record_counts(app, now) + page = await (await admin.get("/plugins/syslog/patterns")).get_data(as_text=True) + assert "Quota full" in page and "NFS server trouble" in page + assert "metric_name=lines_matching_quota_full" in page + assert 'action="/plugins/syslog/patterns/preset/nfs"' not in page # added: no longer offered + rule_form = await (await admin.get( + f"/alerts/rules/new?source_module=syslog&resource_name={src_name}&metric_name=lines_matching_nfs" + )).get_data(as_text=True) + assert '