monitor.collector¶

MetricCollector — collects metrics from multiple UnixHosts via asyncio.gather().

On each tick all hosts are polled simultaneously so results share one timestamp.

Live collection only: a collector polls hosts via asyncio.gather() per tick, optionally persisting to a session-bound MetricDB (schema v2). Review (historical) data is a separate concern — see otto.monitor.export (build_db_export/build_live_export) for the format:1 producer that reads/wraps collected data, and otto.cli.monitor for how a saved export is loaded back for review. A bare MetricCollector with no live targets (e.g. MetricCollector(targets=[])) still declares its parser catalog so get_meta_model() remains well-formed.

class otto.monitor.collector.MetricView(*args, **kwargs)¶

Bases: Protocol

The presentation surface a series needs to be charted.

Both MetricParser and SnmpMetric satisfy this structurally, so the record/publish path is identical whether a point came from a shell command or an SNMP OID.

The members are read-only properties, not plain attributes: the collector only ever reads them, and a mutable attribute member would demand writes that SnmpMetric (frozen=True) cannot accept.

property chart : str¶: Title of the chart this series is drawn on.

property y_title : str¶: Label for the chart’s y-axis.

property unit : str¶: Unit the values are expressed in (e.g. %, MiB).

property tab : str¶: Id of the dashboard tab the chart belongs to.

property tab_label : str¶: Human-readable name of that dashboard tab.

class otto.monitor.collector.MonitorTarget(host: RemoteHost, parsers: dict[str, ~otto.monitor.parsers.MetricParser] = <factory>, core_count: int = 1, snmp: ~otto.monitor.snmp.SnmpSource | None = None)¶

Bases: object

Pairs a UnixHost with the parser dict to use when collecting from it.

By default all hosts use DEFAULT_PARSERS. Pass a custom dict to add new metrics or override built-in commands for a specific host:

MonitorTarget(
    host=gpu_host,
    parsers={
        **DEFAULT_PARSERS,
        "nvidia-smi --query-gpu=utilization.gpu ...": NvidiaGpuParser(),
    },
)

In most cases you do not construct these directly — use register_host_parsers() from an init module and let the CLI build targets automatically.

Set snmp to collect from this host over SNMP instead of by running shell commands — the host then needs no shell parsers, and parsers is ignored. This is what lets otto monitor a host (embedded or Unix) over a channel separate from command execution.

host : RemoteHost¶

parsers : dict[str, MetricParser]¶

core_count : int = 1¶

snmp : SnmpSource | None = None¶

class otto.monitor.collector.MetricCollector(hosts: collections.abc.Sequence[RemoteHost] | None = None, parsers: list[MetricParser] | None = None, db: MetricDB | None = None, targets: list[MonitorTarget] | None = None)¶

Bases: object

Collects numeric metrics from multiple UnixHosts and stores time-series data.

On each tick, all hosts are polled simultaneously via asyncio.gather() so that results from every host share the same timestamp.

Series keys have the form "hostname/metric_label" (e.g. "router1/CPU %").

Parameters:¶

hosts – Remote hosts to monitor. Pass None (or omit) when loading historical data.
parsers – Metric parser instances. Defaults to DEFAULT_PARSERS (cpu, mem, disk, load).
db – Optional session-bound MetricDB for persistence (unopened — this collector opens it lazily via init_db()). None means in-memory only. The collector itself stays session-blind: framing (choosing the label/note and constructing the MetricDB) is the caller’s job, not the collector’s — one process run is one live session.

async init_db() → None¶

Open the persistent DB (no-op without a db at construction). See MetricDB.

Must be awaited before any DB writes. Called automatically by run(); callers that skip run() (e.g. tests) should call this explicitly.

async close_db() → None¶: Close the persistent DB connection and release the file lock.

async close() → None¶

Close the DB connection and all live host sessions.

Must be awaited before the surrounding event loop exits — otherwise subprocess transports owned by LocalSession are GC’d after the loop is closed and raise “Event loop is closed” from __del__.

async run(interval: timedelta = datetime.timedelta(seconds=5), duration: timedelta | None = None) → None¶

Collect metrics from all hosts on each tick.

Each host’s collection is bounded by the interval — if a host does not respond in time, it is skipped for that tick and the session is recovered automatically. This prevents a single slow host from blocking collection on all other hosts.

Commands are bucketed by effective interval (parser.interval or interval) and each bucket runs its own collection loop concurrently, so a parser that declares a faster interval ticks more often without speeding up the rest of that host’s commands. SNMP targets always ride the global (interval) bucket. With no per-parser intervals set, there is exactly one bucket and behavior is identical to a single global loop.

Runs inside the caller’s event loop. Cancel the task wrapping this coroutine (or cancel it directly) to stop collection gracefully.

Parameters:¶

interval – Collection interval as a timedelta.
duration – Optional total run time. None means run forever.

async add_event(label: str, timestamp: datetime | None = None, color: str = '#888888', dash: str = 'dash', source: str = 'manual', end_timestamp: datetime | None = None) → MonitorEvent¶: Record a labeled event and push it to all dashboard subscribers.

async delete_event(event_id: int) → bool¶: Remove an event by id. Returns True if found and removed, False otherwise.

async update_event(event_id: int, label: str, color: str, dash: str, end_timestamp: datetime | None = None) → MonitorEvent | None¶: Update an existing event’s label, color, dash, and end_timestamp. Returns the updated event or None.

get_series() → dict[str, list[MetricPoint]]¶

Return a snapshot of all series (metrics and per-process).

Format: {"hostname/label": [MetricPoint(ts, value, meta), ...]}

get_chart_map() → dict[str, str]¶

Return a mapping of series label → chart group key.

Used by the dashboard to assign historical series (loaded from the monitor_sessions/SSE wire) to their correct chart containers without requiring series_labels() on each parser. The map is built lazily as data arrives in _process_host_results().

get_events() → list[MonitorEvent]¶: Return all recorded events in chronological order.

get_log_events() → list[dict[str, Any]]¶

JSON-safe log-event rows for the monitor_sessions/SSE wire and export.

Shape per row: {"timestamp", "host", "tab", "fields"} — the LogEventRecord spelling, insertion-ordered per (host, tab) ring.

get_meta_model() → MonitorMeta¶

Return the typed internal meta model (see get_meta for the dict form).

Not served at any endpoint directly — see MonitorMeta.

get_meta() → dict[str, Any]¶: Return metadata for the dashboard (host names, metric labels/units, tabs).

subscribe() → Queue[dict[str, Any]]¶: Register a new SSE subscriber and return its queue.

unsubscribe(q: Queue[dict[str, Any]]) → None¶: Remove q from the SSE subscriber list so it receives no further pushes.