Telemetry¶

Telemetry is the central record of what femforge did: every log record, every run, every stored artifact and every sweep point, kept as hive-partitioned parquet under one directory. A run bundle describes one run; telemetry answers questions across thousands of them, such as every error from last night's sweep or every run that used one mesh, without opening each bundle. It is off until you name a root, and a process with telemetry off never imports polars.

Turn it on¶

Name a telemetry root, the directory the tables live under, in one of two ways:

export FEMFORGE_TELEMETRY=telemetry

With a root, the driver appends a row to the run index when it first writes a run's manifest and another once the run settles, so a run is visible while it solves. Every record logged during evaluate() lands in that driver's root. A failed telemetry write only logs a warning; it never fails a solve.

What the root holds¶

telemetry/
├── logs/date=2026-10-13/process=gap-sweep/part-<host>-<pid>-<utc>-<seq>.parquet
├── runs/month=2026-10-01/part-...parquet
├── artifacts/kind=elmer_mesh/part-...parquet
├── sweeps/month=2026-10-01/part-...parquet
├── sweep_events/sweep_id=gap-study/part-...parquet
└── live/logs-<host>-<pid>.jsonl

Telemetry addresses a root and hands out one HiveStore per table. Each table's columns are a polars.Schema whose names come from a StrEnum, such as RunColumn and LogColumn, so you filter with pl.col(RunColumn.STATUS) instead of a bare string.

The store is append-only. Each writer writes its own shard, named by host, process id, a UTC stamp and a sequence number, staged under a temporary name and moved into place, so a reader never sees half a file and many processes write at once without locks. An index re-read adds a new row for the same run; readers keep the latest row per run.

The convergence log is the exception: each bundle keeps its own convergence.jsonl, and scan_convergence reads it as a typed frame with two derived columns, first_relc and progress, the fraction of the way to the tolerance the sif set.

The journal¶

Every log record and every sweep event goes first to the writing process's own JSONL journal under live/, one flushed line per record. The log sink checkpoints the journal into parquet every 500 records, every 5 s, and at exit, then truncates it, so a record sits in exactly one of the two places. A process that dies leaves its journal behind, and the next sink or sweep writer that starts on the root folds it into parquet.

Readers union both halves. scan_logs and SweepQuery.events() return the settled shards together with what is still only in a journal, so a live sweep shows its newest records.

Read it back¶

The query classes are fluent: each method returns a new query, and nothing reads parquet until frame() or page(). This lists the last week's finished runs:

from datetime import datetime, timedelta, timezone

import femforge as ff
from femforge.telemetry import RunColumn, RunIndex

telemetry = ff.Telemetry("telemetry")
index = RunIndex(telemetry, "runs")
added = index.update()

week_ago = datetime.now(timezone.utc) - timedelta(days=7)
recent = index.query().status("ok", "diverged").since(week_ago).newest(20).frame()

print(f"{added} bundles indexed")
print(recent.select(RunColumn.RUN_ID, RunColumn.STATUS, RunColumn.CREATED_AT, RunColumn.WALL_TIME))

update() reads only the bundles whose manifest.json changed since the last pass; update(rebuild=True) reads them all. It parses a dozen manifest keys without validating the config, so a corrupt bundle costs one null instead of the pass.

with_live() joins the claim each run holds right now, read from <run_id>.lock/owner.json, with its heartbeat age. RunQuery.runs() and ArtifactQuery.records() open the matching bundles and artifacts, skipping the ones retention has removed.

For anything the filters do not cover, scan a table as a polars LazyFrame:

import femforge as ff
import polars as pl
from femforge.telemetry import LogColumn

telemetry = ff.Telemetry("telemetry")

errors = (
    telemetry.logs.scan()
    .filter(pl.col(LogColumn.LEVEL) == "ERROR")
    .group_by(LogColumn.PROCESS)
    .len()
    .collect()
)
print(errors)

scan() reads the settled shards only; use scan_logs(telemetry) to include the live journals.

Record a sweep¶

A sweep is the caller's: you pick its id, and a SweepJournal records its launch row and each point transition. Inside sweep_point, every log record and every run index row the driver writes carries the sweep id and the point key:

import traceback

import femforge as ff
from femforge.telemetry import SweepJournal, SweepState, sweep_point


def gap_config(gap_mm: int) -> ff.SimulationConfig:
    """The actuator config at one air gap."""
    return ff.SimulationConfig(
        mesh=ff.MeshConfig(step=f"models/actuator-gap-{gap_mm}mm.step", global_mesh_size=0.002),
        domain=ff.DomainConfig(materials=[ff.materials.Air(), ff.materials.N52()]),
        meta=ff.RunMeta(description=f"Gap {gap_mm} mm", tags={"study": "gap"}),
    )


telemetry = ff.Telemetry("telemetry")
driver = ff.Driver(runs_dir="runs", telemetry=telemetry.root)
gaps = [1, 2, 3]

journal = SweepJournal(telemetry, sweep_id="gap-study")
journal.start("Air gap sweep", total=len(gaps))

for gap in gaps:
    key = f"gap={gap}"
    journal.event(key, SweepState.RUNNING, point_name=f"Gap {gap} mm")
    try:
        with sweep_point(journal.sweep_id, key):
            _ = driver.evaluate(gap_config(gap))
    except Exception:
        journal.event(key, SweepState.FAILED, error=traceback.format_exc())
        continue

    journal.event(key, SweepState.DONE)

journal.finish()

Each worker process of a parallel sweep opens a SweepJournal of its own with the same sweep id. SweepQuery(telemetry, "gap-study").summary() returns a SweepSummary: the count per state, the points settled per hour over the last 30 minutes, or over the sweep's life when it is younger, and the ETA that rate gives.

Follow a run live¶

The workbench's live monitors use all three.

Logging¶

femforge logs through loguru. setup_logging installs the terminal sink and routes the standard library's logging and warnings through it; ff.log runs it with the defaults before its first record.

from pathlib import Path

import femforge as ff
from femforge.common import setup_logging

setup_logging(level="DEBUG", telemetry=Path("telemetry"), process="gap-sweep")

ff.log.bind(gap=0.002, mesh_size=0.0005).info("Gap set.")

The parquet sink takes every record at DEBUG whatever the terminal level, so the store keeps what the terminal hides. Each run also writes its own records, at DEBUG and above, to run.log in its bundle.

Log a short, fixed message and put the data in log.bind(), keyed by field name in snake case without a unit suffix. The terminal prints the bound fields beside the message, and the sink stores them as JSON in the extra column, so you can search the message as text and filter on the fields. run_id, sweep_id and point_key get columns of their own.

Commands¶

The femforge command reads and maintains the store; each subcommand takes --telemetry, else reads FEMFORGE_TELEMETRY.

femforge artifacts ls reads the artifact index when a root resolves and the index has rows, and scans the store otherwise. CLI lists every option.