Distributed Telemetry#

Monarch distributed telemetry collects actor-system events and OpenTelemetry metrics on each host and makes them queryable as one logical database. It uses Apache DataFusion, an extensible SQL query engine built on Apache Arrow. Distributed telemetry is the query layer behind the Monarch Dashboard and complements the live Mesh Admin TUI.

Start with the observability overview if you are choosing a diagnostic surface or configuring a job for the first time.

Quick start#

Enable telemetry before obtaining job state:

from monarch.job import ProcessJob, TelemetryConfig

job = ProcessJob({"workers": 2}).enable_telemetry(
    TelemetryConfig(retention_secs=3600)
)
state = job.state(cached_path=None)

client = state.query_engine_client
if client is None:
    raise RuntimeError("telemetry did not start")

result = client.query(
    """
    SELECT a.full_name, s.new_status, s.reason, s.timestamp_us
    FROM actor_status_events AS s
    JOIN actors AS a ON a.id = s.actor_id
    ORDER BY s.timestamp_us DESC
    LIMIT 50
    """
)
for row in result["rows"]:
    print(row)

QueryEngineClient.query() accepts DataFusion SQL and returns a dictionary whose rows value is a list of dictionaries. Timestamps ending in _us are microseconds since the Unix epoch. IDs are opaque join keys; do not infer meaning from their numeric values.

Core tables#

Table	Contents
`actor_status_events`	Actor status transitions and reasons
`actors`	Actor identity, rank, mesh, and creation time
`events`	Structured tracing events and log fields
`meshes`	Mesh creation, shape, class, name, and parent view
`metric_gauges`	Gauge points with typed `value_f64`, `value_i64`, or `value_u64` values
`metric_histograms`	Explicit histogram counts, typed summaries, bounds, and bucket counts
`metric_sums`	Counter and up-down-counter points with typed sums and sum semantics
`message_status_events`	Message delivery status transitions
`messages`	Individual sender-to-receiver message deliveries
`sent_messages`	One-to-many send operations and destination views
`span_events`	Span enter, exit, and close events
`spans`	Trace span metadata and parent relationships

Mesh-introspection snapshots add actor_failures, actor_inbound_orderings, actor_nodes, children, host_nodes, nodes, ordering_sessions, proc_nodes, resolution_errors, root_nodes, and snapshots. Persisted py-spy captures add pyspy_dumps, pyspy_frames, pyspy_local_variables, and pyspy_stack_traces.

Discover the current table catalog with:

SELECT table_name
FROM information_schema.tables
ORDER BY table_name

Metric tables#

Metrics are split by OTel aggregation rather than numeric type. Query metric_gauges, metric_sums, or metric_histograms based on the instrument aggregation. Each table preserves f64, i64, and u64 in separate nullable columns so large integers do not lose precision.

All metric rows include the instrument name, scope, unit, timestamps, point attributes, and resource attributes. Sum rows also include temporality and monotonicity. Histogram rows retain the count, typed sum, optional typed minimum and maximum, explicit bounds, and bucket counts. Bounds and bucket counts are JSON arrays in bounds_json and bucket_counts_json.

Query examples#

Count actor status transitions:

SELECT new_status, COUNT(*) AS count
FROM actor_status_events
GROUP BY new_status
ORDER BY count DESC

Find recent actor failure events:

SELECT a.full_name, s.reason, s.timestamp_us
FROM actor_status_events AS s
JOIN actors AS a ON a.id = s.actor_id
WHERE s.new_status = 'Failed'
ORDER BY s.timestamp_us DESC
LIMIT 100

Summarize message delivery transitions:

SELECT status, COUNT(*) AS count
FROM message_status_events
GROUP BY status
ORDER BY count DESC

Sum delta counter points within the retention window:

SELECT scope_name, SUM(sum_u64) AS mailbox_posts
FROM metric_sums
WHERE name = 'mailbox.posts'
  AND temporality = 'delta'
  AND sum_u64 IS NOT NULL
GROUP BY scope_name
ORDER BY mailbox_posts DESC

Architecture#

Each producer writes framed Arrow IPC batches to a host-local Unix socket. Trace and entity dispatchers write their corresponding tables, while an OTel UDS metric exporter writes one frame for each non-empty aggregation buffer. A host-local TelemetryActor owns the socket, a Rust DatabaseScanner, and the in-memory tables. The root query engine plans SQL with DataFusion, pushes table scans to active collectors, and merges their Arrow results.

The telemetry job sidecar owns the root collector and query service. Its in-memory state survives repeated state() calls and parent-process refreshes for the same job allocation. Calling job.kill() stops the sidecar and removes that state.

Distributed scans are best-effort. If a worker collector fails or times out, the query can return reduced coverage while the job and healthy collectors continue operating.

Retention and snapshots#

TelemetryConfig.retention_secs applies to sent_messages, messages, message_status_events, metric_gauges, metric_sums, and metric_histograms, spans, span_events, and events. Trace tables are filtered independently by row timestamp, with spans filtered before dependent trace rows. A recent trace row is not displayed by span-joined views if its span has already expired. The default is 3600 seconds. Set it to 0 to disable automatic retention. Periodic maintenance runs approximately once per tenth of the retention window, bounded between 30 seconds and 5 minutes and never longer than the retention window. Queries do not run retention, so expired rows can remain visible until the next periodic sweep. Other core tables have no automatic retention window.

TelemetryConfig.snapshot_interval_secs controls periodic Mesh Admin topology snapshots. The default is 30 seconds; set it to 0 to disable them. Snapshots and event telemetry describe different views of the same job:

Event tables record changes such as actor creation, status transitions, and message delivery.
Snapshot tables capture the administrative topology and actor diagnostics at a point in time.

Troubleshooting#

If state.query_engine_client is None, telemetry bootstrap failed. Check the parent-process logs for job sidecar telemetry bootstrap failed.
If a query is empty immediately after work runs, retry briefly; ingestion and collector flushes are asynchronous.
If rows from one host are missing, inspect collector warnings and use the Mesh Admin TUI to inspect that host.
If a SQL request fails, the raised HTTP error includes the DataFusion parse or planning detail returned by the query service.