Record and query normalized pipeline metrics (the found → mapped → processed funnel).
Emission point: a pipeline calls record/3 at the SAME place it calls
PipelineRun.complete/2 (the sole completer for umbrella/borrowed runs).
Summary
Functions
Short human-readable alert reasons for a dashboard row (empty list ⇒ healthy). Drives
both overall_status/2 and the badge label, so the operator sees why a row is red.
Attach the built-in [:allm_pipeline, :step, :stop] handler that writes
step_logs.queue_time_ms from the emitted queue_time measurement.
Detach the queue_time_ms handler (hot-reload support).
Whether found == 0 should alert for this pipeline (a full-listing scraper).
:telemetry handler for [:allm_pipeline, :step, :stop]: writes
queue_time_ms (the measured backlog wait, converted native→ms) onto the
step-log row named by metadata.step_id.
Recent metrics rows for one pipeline, newest first (for a future trend view).
The most recent metrics row per pipeline_name — the dashboard's primary query.
Uses Postgres DISTINCT ON (pipeline_name) with a matching ORDER BY.
The most recent PipelineRun per run name, as a %{name => run} map — the stale-green
guard's data source. The dashboard joins each metric row to this by pipeline_name: a
hard scrape failure never reaches PipelineRun.complete/2 (so it writes no metric row),
but it DOES leave a :failed run, which this surfaces even though the newest metric is an
old green one. Uses the same Postgres DISTINCT ON idiom as latest_per_pipeline/0, on
pipeline_runs.name. (pipeline_runs is small — one row per run — so the DISTINCT ON
sort is cheap; no new index is required for this query.)
Dashboard health: the metric-intrinsic signals PLUS "the pipeline's latest run
failed/cancelled" (the stale-green guard). last_run is the row from
latest_run_per_pipeline/0 for this pipeline (or nil).
Record one normalized metrics row for run and entity_type. funnel is a map of any
subset of :found/:mapped/:processed/:skipped/:failed/:tokens (absent keys default to 0).
Best-effort: a metrics-write failure must never fail the pipeline run — logs and
returns {:error, changeset} rather than raising.
Metric-intrinsic health of one row: :alert iff failed > 0 (processing trouble)
OR found == 0 for a full-listing scraper (empty scrape — see expects_data?/1).
Types
@type funnel() :: %{ optional(:found) => non_neg_integer(), optional(:mapped) => non_neg_integer(), optional(:processed) => non_neg_integer(), optional(:skipped) => non_neg_integer(), optional(:failed) => non_neg_integer(), optional(:tokens) => non_neg_integer() }
Functions
@spec alert_reasons( ALLM.Pipeline.PipelineMetric.t(), ALLM.Pipeline.PipelineRun.t() | nil ) :: [ String.t() ]
Short human-readable alert reasons for a dashboard row (empty list ⇒ healthy). Drives
both overall_status/2 and the badge label, so the operator sees why a row is red.
@spec attach_step_handler() :: :ok | {:error, :already_exists}
Attach the built-in [:allm_pipeline, :step, :stop] handler that writes
step_logs.queue_time_ms from the emitted queue_time measurement.
The one named consumer of [:allm_pipeline, :step, :stop] (see
ALLM.Pipeline.Telemetry). Idempotent across hot reload — the caller
(a host's Application.start/2) detaches first, since
:telemetry.attach/4 answers {:error, :already_exists} on a duplicate
handler id.
@spec detach_step_handler() :: :ok | {:error, :not_found}
Detach the queue_time_ms handler (hot-reload support).
Whether found == 0 should alert for this pipeline (a full-listing scraper).
The SET is host domain knowledge, not framework knowledge: which pipelines
re-scrape a complete source listing every run — and which are legitimately
empty — is a fact about the host's sources. It is declared as
alert_on_empty: on the host's ALLM.Pipeline.Registry (batch 1.C moved it
off a hardcoded @expects_data_pipelines here) and resolved at runtime by
ALLM.Pipeline.Config.alert_on_empty/0. The reasons for each inclusion and
the one deliberate exclusion travel WITH the values, on the declaration.
Keyed by the run name (= pipeline_metrics.pipeline_name), a string —
not a cron atom; the two namespaces do not line up (extraction plan §3.8a).
Default is OFF for anything undeclared.
:telemetry handler for [:allm_pipeline, :step, :stop]: writes
queue_time_ms (the measured backlog wait, converted native→ms) onto the
step-log row named by metadata.step_id.
Touches only the queue_time_ms column — the step-log structural-identity
property from Phases 1-6 holds for every other column. Best-effort: a write
failure (e.g. a child task with no shared sandbox connection under test) is
logged and swallowed so telemetry never fails a run, and the handler is not
detached by :telemetry for a transient error.
Only fires the write when queue_time is an integer — a plain
(non-fan_out) step emits queue_time: nil and leaves the column untouched.
@spec history(String.t(), pos_integer()) :: [ALLM.Pipeline.PipelineMetric.t()]
Recent metrics rows for one pipeline, newest first (for a future trend view).
@spec latest_per_pipeline() :: [ALLM.Pipeline.PipelineMetric.t()]
The most recent metrics row per pipeline_name — the dashboard's primary query.
Uses Postgres DISTINCT ON (pipeline_name) with a matching ORDER BY.
@spec latest_run_per_pipeline() :: %{ required(String.t()) => ALLM.Pipeline.PipelineRun.t() }
The most recent PipelineRun per run name, as a %{name => run} map — the stale-green
guard's data source. The dashboard joins each metric row to this by pipeline_name: a
hard scrape failure never reaches PipelineRun.complete/2 (so it writes no metric row),
but it DOES leave a :failed run, which this surfaces even though the newest metric is an
old green one. Uses the same Postgres DISTINCT ON idiom as latest_per_pipeline/0, on
pipeline_runs.name. (pipeline_runs is small — one row per run — so the DISTINCT ON
sort is cheap; no new index is required for this query.)
@spec overall_status( ALLM.Pipeline.PipelineMetric.t(), ALLM.Pipeline.PipelineRun.t() | nil ) :: :ok | :alert
Dashboard health: the metric-intrinsic signals PLUS "the pipeline's latest run
failed/cancelled" (the stale-green guard). last_run is the row from
latest_run_per_pipeline/0 for this pipeline (or nil).
@spec record(ALLM.Pipeline.PipelineRun.t(), String.t(), funnel()) :: {:ok, ALLM.Pipeline.PipelineMetric.t()} | {:error, Ecto.Changeset.t() | Exception.t()}
Record one normalized metrics row for run and entity_type. funnel is a map of any
subset of :found/:mapped/:processed/:skipped/:failed/:tokens (absent keys default to 0).
Best-effort: a metrics-write failure must never fail the pipeline run — logs and
returns {:error, changeset} rather than raising.
@spec status(ALLM.Pipeline.PipelineMetric.t()) :: :ok | :alert
Metric-intrinsic health of one row: :alert iff failed > 0 (processing trouble)
OR found == 0 for a full-listing scraper (empty scrape — see expects_data?/1).
This is the helper the per-pipeline Subphase-2 tests use (they hold a metric but no run).
Note unmapped (found - mapped) is deliberately NOT an alert condition — a nonzero
unmapped baseline is normal (e.g. a pipeline that intentionally leaves out-of-scope
items unmapped); it is tracked/displayed for a human to watch, never auto-alerted.