feat(syslog): pattern alerts: telling phrases counted per source each minute for alert rules, storage-trouble presets (#6001)

A pattern (phrases, a severity, a source, a window) is counted every minute
into plugin_metrics as lines_matching_<key> per added source, zero included,
so an ordinary alert rule fires on a match and clears once the window passes.
Each count reads only the lines stored since the last, by line id; a new or
changed pattern reads its window once.

Patterns page with ready-made storage-trouble presets and "Alert on it",
which opens the rule form aimed at the count. Source names are kept unique,
and a rename moves its counts and the rules on them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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)