From 50b89228958fa63315b956124e077706d0b24884 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 30 Oct 2026 14:02:04 -0400 Subject: [PATCH] fix(syslog): a device through Docker's address stays one source when that address changes (#6717) Docker's ingress address depends on the node Steward's task is on, and sources were keyed by (address, sender). After a task move the same device arrived under a new key and became a new "offered" row, so its lines were counted, not kept. A named sender is now one row whatever address it last came from, enforced by a partial unique index. Migration syslog_005 folds rows sharing a name into the added one and moves their lines and patterns across. Co-Authored-By: Claude Opus 5.5 --- docs/plugins/syslog.md | 4 +- .../versions/syslog_005_one_row_per_sender.py | 64 +++++++++++++++++++ plugins/syslog/models.py | 9 ++- plugins/syslog/receiver.py | 43 +++++++++---- tests/integration/test_syslog.py | 46 +++++++++++++ 5 files changed, 150 insertions(+), 16 deletions(-) create mode 100644 plugins/syslog/migrations/versions/syslog_005_one_row_per_sender.py diff --git a/docs/plugins/syslog.md b/docs/plugins/syslog.md index 6beb1b8..5d3aa7e 100644 --- a/docs/plugins/syslog.md +++ b/docs/plugins/syslog.md @@ -56,7 +56,9 @@ network, so the address no longer tells the devices apart. For such an address Steward keys the source by the hostname each line carries instead (#6096): each device is its own source. A device must put its hostname in its lines (NASes and routers do); lines with none at Docker's address all fold into one source, -and the page says so. +and the page says so. Docker's address changes when Steward's task moves to +another node; a named source stays the same source, added or not, and takes the +new address. Such a source's own address, which places its lines on its SNMP device's and host's pages (chart marks, the lines around an anomaly), is found by name: the diff --git a/plugins/syslog/migrations/versions/syslog_005_one_row_per_sender.py b/plugins/syslog/migrations/versions/syslog_005_one_row_per_sender.py new file mode 100644 index 0000000..9b93337 --- /dev/null +++ b/plugins/syslog/migrations/versions/syslog_005_one_row_per_sender.py @@ -0,0 +1,64 @@ +"""Syslog: one row per name at Docker's address, whichever address it last came +from. Rows sharing a name fold into the added one, or else the newest. + +Revision ID: syslog_005_one_row_per_sender +Revises: syslog_004_device_address +Create Date: 2026-10-30 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "syslog_005_one_row_per_sender" +down_revision: Union[str, None] = "syslog_004_device_address" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_STATE_FIRST = {"accepted": 0, "ignored": 1, "offered": 2} + + +def _fold(bind, sender: str) -> None: + rows = bind.execute(sa.text( + "SELECT id, address, state, device_address, host_id, hostname_seen, first_seen_at, last_seen_at, " + "lines_total, last_severity, last_line FROM syslog_sources WHERE sender = :sender" + ), {"sender": sender}).mappings().all() + keep = min(rows, key=lambda r: (_STATE_FIRST.get(r["state"], 3), -r["last_seen_at"].timestamp())) + newest = max(rows, key=lambda r: r["last_seen_at"]) + others = [r["id"] for r in rows if r["id"] != keep["id"]] + in_others = sa.bindparam("others", expanding=True) + for table in ("syslog_lines", "syslog_patterns"): + bind.execute(sa.text(f"UPDATE {table} SET source_id = :keep WHERE source_id IN :others").bindparams(in_others), + {"keep": keep["id"], "others": others}) + bind.execute(sa.text("DELETE FROM syslog_sources WHERE id IN :others").bindparams(in_others), {"others": others}) + bind.execute(sa.text( + "UPDATE syslog_sources SET address = :address, device_address = :device_address, host_id = :host_id, " + "hostname_seen = :hostname_seen, first_seen_at = :first_seen_at, last_seen_at = :last_seen_at, " + "lines_total = :lines_total, last_severity = :last_severity, last_line = :last_line WHERE id = :id" + ), { + "id": keep["id"], + "address": newest["address"], + "device_address": next((r["device_address"] for r in rows if r["device_address"]), ""), + "host_id": keep["host_id"] or next((r["host_id"] for r in rows if r["host_id"]), None), + "hostname_seen": newest["hostname_seen"], + "first_seen_at": min(r["first_seen_at"] for r in rows), + "last_seen_at": newest["last_seen_at"], + "lines_total": sum(r["lines_total"] or 0 for r in rows), + "last_severity": newest["last_severity"], + "last_line": newest["last_line"], + }) + + +def upgrade() -> None: + bind = op.get_bind() + senders = bind.execute(sa.text( + "SELECT sender FROM syslog_sources WHERE sender <> '' GROUP BY sender HAVING count(*) > 1" + )).scalars().all() + for sender in senders: + _fold(bind, sender) + op.create_index("uq_syslog_sources_sender", "syslog_sources", ["sender"], unique=True, + postgresql_where=sa.text("sender <> ''")) + + +def downgrade() -> None: + op.drop_index("uq_syslog_sources_sender", table_name="syslog_sources") diff --git a/plugins/syslog/models.py b/plugins/syslog/models.py index d54858b..924a4e5 100644 --- a/plugins/syslog/models.py +++ b/plugins/syslog/models.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import datetime, timezone from sqlalchemy import (JSON, BigInteger, Boolean, DateTime, ForeignKey, Index, SmallInteger, String, Text, - UniqueConstraint) + UniqueConstraint, text) from sqlalchemy.orm import Mapped, mapped_column from steward.core.ids import time_ordered_id @@ -17,6 +17,8 @@ class SyslogSource(Base): Through a port Swarm publishes on its ingress (the default), every device arrives from one address of Docker's own; there a row is a device by the name its lines carry, `sender` (#6096). Everywhere else `sender` is "". + That address changes when Steward's task changes node, so a name has one + row and `address` follows it. A sender Steward hasn't been told about is a row too, `state="offered"`: its lines are counted and its newest kept as a sample, so the page can @@ -25,7 +27,10 @@ class SyslogSource(Base): on; `"ignored"` stops the offer and keeps counting. """ __tablename__ = "syslog_sources" - __table_args__ = (UniqueConstraint("address", "sender", name="uq_syslog_sources_address_sender"),) + __table_args__ = ( + UniqueConstraint("address", "sender", name="uq_syslog_sources_address_sender"), + Index("uq_syslog_sources_sender", "sender", unique=True, postgresql_where=text("sender <> ''")), + ) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) address: Mapped[str] = mapped_column(String(64), nullable=False) diff --git a/plugins/syslog/receiver.py b/plugins/syslog/receiver.py index 7a6f96a..330de74 100644 --- a/plugins/syslog/receiver.py +++ b/plugins/syslog/receiver.py @@ -23,7 +23,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING -from sqlalchemy import func, select +from sqlalchemy import func, or_, select from steward.core.scheduler import ScheduledTask @@ -246,21 +246,36 @@ async def store_lines(session, lines: list[Received]) -> None: for line in lines: by_sender.setdefault(sender_of(line), []).append(line) addresses = list({address for address, _ in by_sender}) - sources = {(s.address, s.sender): s for s in (await session.execute( - select(SyslogSource).where(SyslogSource.address.in_(addresses)))).scalars()} - new = [k for k in by_sender if k not in sources] - for key in [k for k in new if k[1]]: - # A row from before devices at Docker's address were told apart by - # name, last seen under this one, is this device: it keeps its state - # and name rather than a new row being offered beside it. - before = sources.get((key[0], "")) - if before is not None and before.hostname_seen == key[1]: - before.sender = key[1] - sources[key] = sources.pop((key[0], "")) - new.remove(key) + names = list({name for _, name in by_sender if name}) + rows = (await session.execute(select(SyslogSource).where( + or_(SyslogSource.address.in_(addresses), SyslogSource.sender.in_(names))))).scalars().all() + sources = {(s.address, s.sender): s for s in rows} + by_name = {s.sender: s for s in rows if s.sender} + new = [] + for key in by_sender: + address, name = key + if name in by_name: + # Docker's address depends on the node Steward's task is on, so a + # device is its name wherever it now arrives from. + by_name[name].address = address + sources[key] = by_name[name] + elif key not in sources: + # A row from before devices at Docker's address were told apart by + # name, last seen under this one, is this device: it keeps its state + # and name rather than a new row being offered beside it. + before = sources.get((address, "")) + if name and before is not None and before.hostname_seen == name: + before.sender = name + by_name[name] = sources[key] = sources.pop((address, "")) + else: + new.append(key) if new: known = (await session.execute(select(func.count()).select_from(SyslogSource))).scalar_one() for key in new: + if key[1] in by_name: # the same name from a second address in this round + by_name[key[1]].address = key[0] + sources[key] = by_name[key[1]] + continue if known >= MAX_SENDERS: LISTENER.unoffered += len(by_sender.pop(key)) continue @@ -268,6 +283,8 @@ async def store_lines(session, lines: list[Received]) -> None: source = SyslogSource(address=key[0], sender=key[1], first_seen_at=first.at, last_seen_at=first.at) session.add(source) sources[key] = source + if key[1]: + by_name[key[1]] = source known += 1 await session.flush() diff --git a/tests/integration/test_syslog.py b/tests/integration/test_syslog.py index ebc723e..70698c3 100644 --- a/tests/integration/test_syslog.py +++ b/tests/integration/test_syslog.py @@ -120,6 +120,52 @@ def test_devices_through_dockers_address_are_told_apart_by_name(app, monkeypatch asyncio.run(go()) +def test_a_device_stays_its_source_when_dockers_address_changes(app, monkeypatch): + """Docker's address depends on the node Steward's task is on: an added + device keeps its row, its state and its lines after the task moves, even + when a round carries both addresses.""" + from sqlalchemy import delete, select + + SyslogLine, SyslogSource = _mod("models").SyslogLine, _mod("models").SyslogSource + receiver = _mod("receiver") + parse = _mod("parse").parse + monkeypatch.setattr(receiver, "on_own_network", lambda address: address.startswith("10.0.0.")) + now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc) + + def line(address, host, text): + return receiver.Received(now, address, parse(f"<14>Oct 19 01:12:03 {host} kernel: {text}")) + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(SyslogSource)) + s.add(SyslogSource(address="10.0.0.3", sender="SynNAS", state="accepted", name="Mine", + first_seen_at=now, last_seen_at=now, lines_total=1)) + async with app.db_sessionmaker() as s: + async with s.begin(): + await receiver.store_lines(s, [line("10.0.0.3", "SynNAS", "a"), line("10.0.0.4", "SynNAS", "b")]) + async with app.db_sessionmaker() as s: + async with s.begin(): + await receiver.store_lines(s, [line("10.0.0.4", "SynNAS", "c")]) + async with app.db_sessionmaker() as s: + src = (await s.execute(select(SyslogSource))).scalar_one() + assert (src.address, src.state, src.name, src.lines_total) == ("10.0.0.4", "accepted", "Mine", 4) + kept = (await s.execute(select(SyslogLine.message))).scalars().all() + assert sorted(kept) == ["a", "b", "c"] + # A name first heard from two addresses in one round is still one source. + async with app.db_sessionmaker() as s: + async with s.begin(): + await receiver.store_lines(s, [line("10.0.0.3", "AsusNAS", "d"), line("10.0.0.4", "AsusNAS", "e")]) + async with app.db_sessionmaker() as s: + src = (await s.execute(select(SyslogSource).where(SyslogSource.sender == "AsusNAS"))).scalar_one() + assert (src.address, src.state, src.lines_total) == ("10.0.0.4", "offered", 2) + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(delete(SyslogSource)) + + asyncio.run(go()) + + def test_the_ring_keeps_each_source_within_its_window_and_cap(app): from sqlalchemy import delete, select