# threepm-io

Application metrics, events, **actions and calls** for a process started by
[3pm](https://github.com/Unitech/3pm) — pm2's `io.metric` / `io.counter` /
`io.notify_error`, for 3pm, in Python. Standard library only, Python ≥ 3.9.

```console
pip install --index-url https://3pm.edgecr4ft.net/simple/ threepm-io    # or: pip install -e 3pm/packages/python
```

Published on <https://3pm.edgecr4ft.net> only, not on PyPI. Keep
`--index-url` (a `requirements.txt` takes the same line), not
`--extra-index-url`, which would also let pip pick a same-named package on
PyPI.

```python
import threepm_io as io

io.init(source="cv-engine")                      # one connection per process of the app
xruns   = io.counter("cv.xruns", description="PortAudio over/underruns")
jitter  = io.histogram("cv.callback.jitter", unit="ms")
clients = io.gauge("cv.ws.clients")
io.metric("cv.queue.depth", lambda: len(queue))  # read at every flush, in the flush thread

xruns.inc()
jitter.observe(0.42)
clients.set(3)
io.event("midi.port_lost", level="warn", attrs={"port": "ES-9"})
try:
    open_device()
except Exception as e:
    io.notify_error(e, attrs={"device": "ES-9"})  # an `error` event with the traceback
```

Then `3pm show <app>` (section "Custom metrics") and `3pm events -f app`.

Outside 3pm (no `THREEPM_CTL_SOCK` in the environment) every call is a
no-op. Nothing blocks the caller and nothing raises: a daemon thread
connects, sends the whole registry every second, reconnects with backoff
(1 s → 30 s) and gives up after five refused `hello`s (the process outlived
its incarnation). Events raised before the connection is up are queued
(64, under the same 64 KiB outbound cap). A short-lived script should
`io.flush(wait_ms=500)` before exiting; `shutdown(wait_ms=500)` is that
flush followed by the close.

Cache the series you write to on a hot path (`requests = io.counter(...)`
once, `requests.inc()` per event): `.labels({...})` is a dictionary lookup
plus validation, fine per request, wasteful per sample. Past 50 label sets
a metric folds into one `overflow="true"` series; `.labels()` on that fold
returns the fold itself.

| Call | Notes |
|---|---|
| `init(source=None, flush_ms=1000, on_shutdown=None, connect_wait_ms=5000)` | Idempotent; `source` defaults to `main` (see below); `on_shutdown` runs when the daemon asks for a graceful stop (`shutdown_with_message`), in the reader thread — it may call `shutdown()`. Test knobs, in milliseconds like the Node and C++ SDKs: `reconnect_min_ms`, `reconnect_max_ms`, `max_buffer_bytes`, `max_refusals`, `refusal_window_ms` |
| `gauge` / `counter` / `meter` / `histogram(name, unit=None, description=None, labels=None, scale=1)` | Same name + labels = same object; `.labels({...})` derives a child; counters are cumulative; histograms are base-2 exponential (`scale` 0..4) and never carry more than 128 non-empty buckets — a value that would open a 129th lowers the scale by one (buckets merge pairwise, `count`/`sum` exact) until it fits; past 50 label sets a metric folds into `overflow="true"`. Non-finite numbers (NaN, infinities, integers beyond a double) are ignored |
| `metric(name, fn, ...)` | Gauge by callback — `fn` runs in the flush thread, outside any asyncio loop |
| `event(name, level=None, attrs=None, exc=None, fp=None)` | `level` in `debug`/`info`/`warn`/`error`/`fatal`; `attrs` flat scalars (≤ 16) |
| `notify_error(err, attrs=None, name=None)` | `error`-level event carrying `exc.{type,message,stacktrace}` |
| `flush(wait_ms=None)`, `shutdown(wait_ms=500)` | `flush` waits for the connection then the bytes when `wait_ms` is given (never from the flush thread itself); `shutdown` is `flush(wait_ms)` then the close |

## Actions and calls: `io.action`, `io.call`

A function another process — `3pm trigger`, `3pm call`, an `io.call` from
any SDK, the fleet backend — can call, on this machine or from another
node of the bucket, **with a file in and a file out**. Same wire as the
Node SDK (`packages/node/README.md`, "Fleet calls"), so a Node caller
reaches a Python action and the reverse.

```python
import threepm_io as io
io.init()                                   # source `main` by default, as in Node

@io.action("segment", input={"max_bytes": 50_000_000, "ext": ["png", "jpg"]}, timeout_ms=60_000)
def segment(params, ctx):                   # ctx.input: path, bytes, ext, sha256 — fetched and checked
    return model.run(ctx.input.path, params)            # JSON, ≤ 512 KiB, ≤ 64 levels deep

@io.action("upscale", input=50_000_000, output={"max_bytes": 20_000_000})
def upscale(params, ctx):
    render(ctx.input.path, ctx.file + ".tmp")           # a result too big for the line is a file
    ctx.commit_file("png")                              # renamed; PUT to the caller when it asked for it
    return {"seed": params.get("seed")}

io.serve()                                  # a worker that only answers calls: block until 3pm stops it
```

```python
res = io.call("gpu-1", "vision", "segment", {"threshold": 0.7}, file="/tmp/capture.png")
res["data"]                                 # {call_id, ok, data, duration_ms, results}
up = io.call("gpu-1", "vision", "upscale", file="in.png", output="/tmp/out.png")
up["file"]                                  # {"path": "/tmp/out.png", "bytes": …, "ext": "png"}
io.call(None, "vision", "segment", file=f)  # same machine: by path, neither agent nor backend
```

| Call | Notes |
|---|---|
| `action(name, handler=None, *, description, remote=True, timeout_ms, input, output, concurrent)` | `handler(params, ctx)` runs **in a thread of its own** (a coroutine function is run to completion there), so a slow handler never stalls the connection; what it raises is the failure (`"ValueError: …"`, cut to 1 KiB). Returns `unregister`; without `handler` it is a decorator. `input` is `{"max_bytes": n, "ext": [...]}` or a bare size; without it a call carrying bytes is refused by the daemon. `concurrent` (1..16) is how many calls of that action run at once; past it a call is `busy`. A bad name, a `3pm:` name, a 33rd action or an unreadable `output` is logged (`THREEPM_IO_DEBUG=1`) and ignored |
| `ctx` | `id`, `name`, `timeout_ms`, `deadline_epoch_ms`, `input` (`None` without bytes), `file`, `commit_file(ext)` → size, `committed`, `output`. An input blob is staged next to `ctx.file`, checked against the size and digest the call declared **before** the handler runs (a mismatch fails the call), and deleted once the reply is out; a local call's input is the caller's own file, never deleted. `commit_file` raises — and deletes the `.tmp` — past the call's deadline, and with `too_large` past the declared `output.max_bytes` |
| `call(node, app, action, params=None, *, file, ext, sha256, bytes, output, instance=0, source, timeout_ms)` | **Blocks** until the answer and **raises `CallError`** (`e.code`: `timeout`, `refused`, `too_large`, `busy`, `gone`, `failed`, `lost`, `node_offline`, `node_unknown`, `ambiguous`, `forbidden`, `agent_too_old`, `daemon_too_old`, or a newer one passed through; `str(e)` is `code: message`). `node` is a `3pmn_…` id or a unique hostname, `None` this machine. `sha256` + `bytes` skip the hashing pass when the digest is already held. `output` is a path to bring the action's file to, or `True` to be told the blob only. `instance=None` asks every instance and is refused with a `file`. At most 16 calls in flight per connection. Outside 3pm: `node_offline` |
| `ready(timeout_ms=None)` | Wait for the control connection (`connect_wait_ms`, 5 s, by default — what `call` does first); `False` outside 3pm |
| `serve()` | Block until the daemon's `shutdown` or `io.shutdown()`; returns at once outside 3pm. Nothing else keeps a Python process up: the SDK's threads are daemon threads |

The bytes never ride the control socket: they are `PUT`/`GET` on the
fleet agent's Unix socket (`$THREEPM_DATA_SOCK`, else
`agent/data-<uid>.sock` then `agent/data.sock` under the 3pm home, else
next to the control socket), with the token of the call's ticket. No
agent on the host: a call with a `file` to another node is
`node_offline` before it is placed; the local call still works.

**Not in the Python SDK yet:** streams (`io.open` / `io.stream`), a
streamed input (`input.stream`), the built-in `3pm:profile`.

**`init()` without `source` now names itself** — `main` for the process
3pm spawned, `pid-<pid>` for a process that one started (the
`THREEPM_IO_MAIN` mark it inherits) — as the Node SDK does: a sourceless
connection can report metrics but declare no action. Series of an app
that passed no `source` gain `src="main"`.

`threepm_io.Io()` gives an independent instance (another `source` in the
same process). Full reference, bounds and the other languages:
`docs/app-metrics.md` in the 3pm repository. Debug trace:
`THREEPM_IO_DEBUG=1`.