bvandeusenandClaude Opus 5.5
45aae6d585
Patterns could only be added on the Syslog page, so an agent asked to map them on the NASes couldn't. list_syslog_patterns gives each pattern's metric and the ready-made ones not added; add_syslog_preset (admin, audited) adds one. The key-picking helper moves beside the patterns for both callers. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
210 lines
11 KiB
Python
210 lines
11 KiB
Python
"""The syslog plugin's MCP tools (#6000): the sources and the lines kept from
|
|
them. Registered through register_mcp_tools() in __init__.py, so they exist
|
|
only while the plugin is enabled. They read what the receiver stored."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from steward.core.paging import keyset
|
|
from steward.mcp.tools._audit import audit_mcp
|
|
from steward.mcp.tools._shape import age_s, clamp_limit, cut_middle, db_session, iso, lookback, now_utc, pick
|
|
from steward.models.hosts import Host
|
|
from steward.models.users import UserRole
|
|
|
|
from .models import SyslogLine, SyslogPattern, SyslogSource
|
|
from .parse import facility_name, severity_name
|
|
from .patterns import PRESETS, PRESETS_BY_KEY, SOURCE_MODULE, free_key, resource_of
|
|
from .service import (AROUND_LINES, AROUND_SEVERITY, LineFilter, addresses_by_name, device_address, filtered,
|
|
lines_between, place_for, rates, severity_level, shown_on, source_name, sources_for)
|
|
|
|
|
|
async def list_syslog_sources(state: str = "") -> dict:
|
|
"""The devices sending syslog to Steward.
|
|
|
|
state: "accepted" (added: their lines are kept and searchable),
|
|
"offered" (sending but not added: counted, not kept), "ignored";
|
|
empty for all.
|
|
|
|
A device whose lines reach Steward through Docker's own address (a port
|
|
published on Swarm's ingress) is told apart by the name its lines carry:
|
|
`through_docker` is true and `address` is Docker's, shared with the others.
|
|
`device_address` is the device's own: set on the source, or found from the
|
|
SNMP device or host its name is (else the address its lines come from).
|
|
|
|
Each gives its address, the host it is shown on, lines received in all
|
|
(`lines_total`) and, for an added source, in the last hour, when it was
|
|
last heard and its newest line. search_syslog searches the kept lines.
|
|
"""
|
|
if state and state not in ("accepted", "offered", "ignored"):
|
|
raise ValueError("state must be accepted, offered or ignored, or empty for all.")
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
stmt = select(SyslogSource).order_by(SyslogSource.last_seen_at.desc())
|
|
if state:
|
|
stmt = stmt.where(SyslogSource.state == state)
|
|
sources = list((await db.execute(stmt)).scalars())
|
|
per_hour = await rates(db, [s.id for s in sources if s.state == "accepted"], now)
|
|
hosts = {h.id: h.name for h in (await db.execute(select(Host))).scalars()}
|
|
named = await addresses_by_name(db) if any(s.sender for s in sources) else None
|
|
return {"sources": [{
|
|
"source_id": s.id, "name": source_name(s), "address": s.address, "state": s.state,
|
|
"through_docker": bool(s.sender), "device_address": device_address(s, named),
|
|
"host_id": s.host_id, "host_name": hosts.get(s.host_id), "calls_itself": s.hostname_seen or None,
|
|
"lines_total": s.lines_total, "lines_last_hour": per_hour.get(s.id, 0) if s.state == "accepted" else None,
|
|
"last_heard": iso(s.last_seen_at), "last_heard_age_s": age_s(s.last_seen_at, now),
|
|
"newest_line": {"severity": severity_name(s.last_severity) if s.last_severity is not None else None,
|
|
"text": cut_middle(s.last_line)},
|
|
} for s in sources]}
|
|
|
|
|
|
async def search_syslog(source: str = "", host: str = "", text: str = "", severity: str = "",
|
|
app: str = "", range: str = "1h", before: str = "", limit: int = 50) -> dict:
|
|
"""Search the syslog lines Steward has kept from added sources, newest
|
|
first: a NAS's, a router's or another device's own log.
|
|
|
|
source: a source's name, address or id (list_syslog_sources lists them);
|
|
empty searches every added source.
|
|
host: a host's name or id: the sources shown on that host (bound to it,
|
|
or at its address). Empty for all.
|
|
text: case-insensitive substring of the message; empty matches every line.
|
|
severity: this level and worse: "err" (errors, crit, alert, emerg),
|
|
"warning", "notice", "info", "debug", or 0-7. Empty for all.
|
|
app: the app or tag that wrote the line (e.g. "kernel", "sshd"), exact.
|
|
range: how far back — "1h" (default), "6h", "24h", "7d" or "30d". Each
|
|
source keeps its lines up to a size and age limit, so older ones may be gone.
|
|
before: an ISO time; only lines received before it. Pass a response's
|
|
`next_before` to page further back.
|
|
limit: lines to return, default 50, max 200.
|
|
|
|
`at` is when Steward received the line, the time to trust; `sent_at` is
|
|
the device's own clock, given only when the line carried its zone.
|
|
"""
|
|
_, key = lookback(range)
|
|
limit = clamp_limit(limit, 50, 200)
|
|
f = LineFilter(severity=severity_level(severity), app=app.strip(), text=text.strip(), range=key)
|
|
before_dt = None
|
|
if before:
|
|
try:
|
|
before_dt = datetime.fromisoformat(before.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise ValueError(f"before {before!r} is not an ISO-8601 time; pass a response's next_before.") from exc
|
|
async with db_session() as db:
|
|
accepted = list((await db.execute(
|
|
select(SyslogSource).where(SyslogSource.state == "accepted"))).scalars())
|
|
if not accepted:
|
|
return {"range": key, "matches": [], "returned": 0, "truncated": False,
|
|
"note": "No syslog source is added yet; list_syslog_sources shows who is sending."}
|
|
chosen = accepted
|
|
if source:
|
|
want = source.strip()
|
|
by_address = [s for s in accepted if s.address == want]
|
|
chosen = by_address or [pick(want, accepted, kind="syslog source",
|
|
list_tool="list_syslog_sources", name=source_name)]
|
|
if host:
|
|
h = pick(host, list((await db.execute(select(Host))).scalars()), kind="host", list_tool="list_hosts")
|
|
chosen = sources_for(chosen, h.id, h.address, await addresses_by_name(db))
|
|
if not chosen:
|
|
return {"range": key, "matches": [], "returned": 0, "truncated": False,
|
|
"note": f"No added syslog source is shown on {h.name}."}
|
|
names = {s.id: source_name(s) for s in chosen}
|
|
stmt = filtered(select(SyslogLine).where(SyslogLine.source_id.in_(list(names))), f)
|
|
if before_dt is not None:
|
|
stmt = stmt.where(SyslogLine.received_at < before_dt)
|
|
rows = list((await db.execute(
|
|
keyset(stmt, SyslogLine.received_at, SyslogLine.id, None, size=limit))).scalars())
|
|
more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
out = {"range": key, "returned": len(rows), "truncated": more, "matches": [{
|
|
"at": iso(r.received_at), "sent_at": iso(r.sent_at), "source": names[r.source_id],
|
|
"severity": severity_name(r.severity), "facility": facility_name(r.facility),
|
|
"app": r.app or None, "hostname": r.hostname or None, "message": cut_middle(r.message),
|
|
} for r in rows]}
|
|
if more:
|
|
# Full precision: `at` is cut to the second, and a cursor cut the same
|
|
# way would skip lines received earlier in that second.
|
|
out["next_before"] = rows[-1].received_at.isoformat()
|
|
out["more"] = "More lines are older; pass next_before as `before`, or narrow with `text` or `severity`."
|
|
return out
|
|
|
|
|
|
async def list_syslog_patterns() -> dict:
|
|
"""The phrases and severities counted in the kept syslog lines, and the
|
|
ready-made ones not added yet.
|
|
|
|
Each pattern's count of matching lines over its window is the metric
|
|
`metric` of source module `source_module`, per added source (`resources`,
|
|
empty until a source is added), written every minute with zeros included.
|
|
An alert rule on it ("> 0") fires on the first match and clears once the
|
|
window has passed. add_syslog_preset adds a ready-made one.
|
|
"""
|
|
async with db_session() as db:
|
|
patterns = list((await db.execute(select(SyslogPattern).order_by(SyslogPattern.name))).scalars())
|
|
accepted = list((await db.execute(
|
|
select(SyslogSource).where(SyslogSource.state == "accepted"))).scalars())
|
|
taken = {p.key for p in patterns}
|
|
return {"source_module": SOURCE_MODULE, "patterns": [{
|
|
"key": p.key, "name": p.name, "metric": p.metric, "phrases": p.phrases or [],
|
|
"severity_and_worse": severity_name(p.max_severity) if p.max_severity is not None else None,
|
|
"window_minutes": p.window_minutes, "enabled": p.enabled,
|
|
"resources": [resource_of(s) for s in accepted if p.source_id in (None, s.id)],
|
|
} for p in patterns], "presets_not_added": [{
|
|
"key": p.key, "name": p.name, "about": p.about,
|
|
} for p in PRESETS if p.key not in taken]}
|
|
|
|
|
|
async def add_syslog_preset(key: str) -> dict:
|
|
"""Add a ready-made pattern, counted for every added source. The same as
|
|
"Add" under Syslog → Patterns. Disk I/O errors, disk resets, NFS server
|
|
trouble, filesystem errors, degraded arrays and errors and worse:
|
|
list_syslog_patterns gives their keys. Recorded in the audit log as
|
|
"<you> via MCP".
|
|
"""
|
|
preset = PRESETS_BY_KEY.get((key or "").strip())
|
|
if preset is None:
|
|
raise ValueError(f"No preset {key!r}. Presets: {', '.join(p.key for p in PRESETS)}.")
|
|
async with db_session() as db, db.begin():
|
|
if (await db.execute(select(SyslogPattern).where(SyslogPattern.key == preset.key))).scalars().first():
|
|
raise ValueError(f"{preset.name!r} is already added.")
|
|
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()
|
|
added = {"pattern_id": pattern.id, "key": pattern.key, "name": pattern.name, "metric": pattern.metric}
|
|
await audit_mcp("syslog.pattern_added", "syslog_pattern", preset.name, {"preset": preset.key})
|
|
return added
|
|
|
|
|
|
def _as_events(lines: list[SyslogLine], sources: list[SyslogSource]) -> list[dict]:
|
|
from steward.mcp.tools._events import event
|
|
names = {s.id: source_name(s) for s in sources}
|
|
return [event(ln.received_at, "syslog",
|
|
f"{severity_name(ln.severity)}{' ' + ln.app if ln.app else ''}: {cut_middle(ln.message, 300)}",
|
|
subject=names[ln.source_id]) for ln in reversed(lines)]
|
|
|
|
|
|
async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]:
|
|
"""Warnings and worse in a window (get_anomaly) from the syslog sources
|
|
shown on a host: bound to it, or at its address."""
|
|
sources = await shown_on(db, host_id, None)
|
|
return _as_events(await lines_between(db, sources, since, until, AROUND_SEVERITY, AROUND_LINES), sources)
|
|
|
|
|
|
async def mcp_resource_events(db, source: str, resource: str, since, until) -> list[dict]:
|
|
"""The same for an SNMP device, which has no host to read events by: the
|
|
sources at its address, or bound to its host."""
|
|
if source != "snmp":
|
|
return []
|
|
host_id, address = await place_for(db, source, resource)
|
|
if not (host_id or address):
|
|
return []
|
|
sources = await shown_on(db, host_id, address)
|
|
return _as_events(await lines_between(db, sources, since, until, AROUND_SEVERITY, AROUND_LINES), sources)
|
|
|
|
|
|
def register_tools(register) -> None:
|
|
for fn in (list_syslog_sources, search_syslog, list_syslog_patterns):
|
|
register(fn, role=UserRole.viewer, read_only=True)
|
|
register(add_syslog_preset, role=UserRole.admin, read_only=False)
|