# 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](https://github.com/apache/datafusion), an extensible SQL
query engine built on [Apache Arrow](https://arrow.apache.org/). Distributed
telemetry is the query layer behind the [Monarch Dashboard](monarch-dashboard)
and complements the live [Mesh Admin TUI](admin-tui).

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

## Quick start

Enable telemetry before obtaining job state:

```python
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:

```sql
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:

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

Find recent actor failure events:

```sql
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:

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

Sum delta counter points within the retention window:

```sql
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](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.

## Related documentation

- [Metrics](metrics)
- [Observability overview](observability)
- [Monarch Dashboard](monarch-dashboard)
- [Mesh Admin TUI](admin-tui)
- [OpenTelemetry and Grafana](./generated/examples/otel_collector)
- [Jobs API](api/monarch.job)