monitor.log_sourced¶

Log-sourced metric parsers — data-carried timestamps from files on the host.

Some systems don’t expose live values: a cron job digests performance numbers into a timestamped CSV file every few minutes, or the interesting record is a log file’s event stream. Both ride the existing shell acquisition path via parse_tick() — the command IS the reduction step (cat/tail/awk/grep/jq on the host ships back only the lines otto needs; the design assumes source data is always textually reducible on the host).

from otto.monitor.log_sourced import CsvMetricParser
from otto.monitor.parsers import register_parsers

register_parsers(
    [
        CsvMetricParser(
            "cat /var/log/perf/net.csv",
            columns=["rx_kbps", "tx_kbps"],
            chart="Cron net digest",
            tab="network",
            tab_label="Network",
            unit="kb/s",
            interval=60,
        ),
    ]
)

Timestamp convention: naive values are treated as UTC.

This module is never imported by otto’s eager import chain (import-budget guard) — import it explicitly from init modules or test code.

class otto.monitor.log_sourced.ProvisionalTail¶

Bases: object

Hold back the final line of each read until a later read confirms it.

The shell transport strips trailing newlines before parsers see output, so a torn (mid-write) final line is indistinguishable from a complete one by text alone — and a torn line can still parse, which would poison the high-water mark and persist a wrong row. Unless the output still carries its final newline (then the last line is provably complete), the final line is provisional: it emits once a subsequent read shows it unchanged, with or without newer lines after it. Worst-case latency for the newest row is one poll interval; a torn line never emits (its completed form replaces it and emits after stabilizing in turn).

lines(output: str) → list[str]¶: Split output into lines, holding back an unconfirmed final line.

otto.monitor.log_sourced.parse_timestamp(text: str, fmt: str = 'auto') → datetime | None¶

Parse a data-carried timestamp; None (skip the row) when it doesn’t parse.

fmt:

"auto": epoch seconds, else ISO-8601 (the CSV first-column convention);
"epoch": Unix epoch seconds (int or float);
"iso": ISO-8601 (a Z suffix is accepted on Python 3.10);
anything else: a strptime format. For a format without a year directive (classic syslog), the current UTC year is injected before parsing; if that lands the result more than 2 days in the future (a “Dec 31” line read just after New Year), one year is subtracted.

Naive results are treated as UTC in every mode.

class otto.monitor.log_sourced.HighWaterMark¶

Bases: object

Timestamp high-water mark: makes re-reads of rolling files idempotent.

Tracks the newest emitted row timestamp per parser instance (parser instances are per-target deep copies, so state never leaks across hosts). Rows at or below the mark were already emitted on a previous tick and are dropped; survivors come back sorted ascending and the mark advances to the newest survivor. Keyed on row timestamps, not file offsets, so log rotation/truncation needs no special handling — new rows are still newer.

The final line of each read is held back by the parser BEFORE this filter (see ProvisionalTail) until a later read confirms it unchanged, so a torn (mid-write) last line never reaches the mark — only its stabilized, completed form does. Boundary rule: a new row bearing exactly the mark’s timestamp is dropped as already-seen (accepted trade-off; real sources have per-row-unique or sub-second timestamps).

advance(rows: Iterable[tuple[datetime, T]]) → list[tuple[datetime, T]]¶: Return only the rows newer than the mark, ascending; advance the mark.

class otto.monitor.log_sourced.CsvMetricParser(command: str, columns: Sequence[str], *, chart: str, tab: str = 'metrics', tab_label: str = 'Metrics', y_title: str = '', unit: str = '', interval: float | None = None)¶

Bases: MetricParser

Chart metrics from a cron-digested CSV file read over the shell.

Line format: first column an ISO-8601 or epoch-seconds timestamp (naive = UTC), remaining columns numeric values matching columns (the series labels), comma-separated. Header, torn, and otherwise malformed lines are skipped — a mid-write read self-heals next tick because the high-water mark never passes a skipped line. Points carry their data-carried timestamps, so a file holding the last hour backfills the dashboard and DB with an hour of real history on monitor start.

One instance per file: the command string is the parser registry key (“a couple of CSV files” = two registered instances). Register via register_parsers() or register_host_parsers().

Parameters:¶

command – Shell command printing the CSV content (e.g. cat /var/log/perf/net.csv).
columns – Series label per value column, in file order (timestamp column excluded).
chart – Chart group id the series render on (one chart per parser).
tab – Dashboard tab id.
tab_label – Human-readable tab button label.
y_title – Y-axis title shown left of the chart.
unit – Unit suffix for chart annotations.
interval – Poll cadence override in seconds (e.g. 60 for a file written every 5 minutes).

parse(output: str, *, ctx: ParseContext) → dict[str, MetricDataPoint]¶: Unused — this parser produces timed samples via parse_tick().

parse_tick(output: str, *, ctx: ParseContext) → TickResult¶

Convert one tick’s command output into timed samples and/or log events.

The collector always calls this, never parse() directly. The default wraps parse() as a single untimed sample, so existing parsers behave exactly as before. Log-sourced parsers (see otto.monitor.log_sourced) override this to emit data-carried timestamps — multiple samples per read, ascending — and/or columnar LogEvent rows.

class otto.monitor.log_sourced.RegexLogEventParser(command: str, pattern: str | Pattern[str], *, tab: str, tab_label: str, ts_group: str = 'ts', ts_format: str = 'iso', interval: float | None = None)¶

Bases: MetricParser

Columnar log-event rows from a log file read over the shell.

Each line matching pattern becomes a table row: the named groups become the columns (table_columns, in pattern order), except ts_group, which carries the row timestamp. Non-matching lines are skipped — a wrong pattern therefore produces zero rows ever, which the collector’s silent-parser warning surfaces by tick 3. Re-reads of the tail -n N window dedup on the row-timestamp high-water mark, so an append-only log of any size fits: the window bounds every read, the mark discards overlap.

This parser contributes a kind="table" dashboard tab (its own — table parsers must not share a tab id with chart parsers) and no chart.

Parameters:¶

command – Shell command printing the log window (e.g. tail -n 200 /var/log/syslog).
pattern – Line regex with named groups; searched per line.
tab – Dashboard tab id for the table.
tab_label – Human-readable tab button label.
ts_group – Name of the group holding the timestamp.
ts_format – "iso", "epoch", or a strptime format (see parse_timestamp()).
interval – Poll cadence override in seconds.

parse(output: str, *, ctx: ParseContext) → dict[str, MetricDataPoint]¶: Unused — this parser produces log events via parse_tick().

parse_tick(output: str, *, ctx: ParseContext) → TickResult¶

Convert one tick’s command output into timed samples and/or log events.