ALLM.Pipeline.Telemetry (allm_pipeline v0.1.0)

Copy Markdown View Source

The [:allm_pipeline, …] telemetry event contract, plus thin emit_* helpers wrapping :telemetry.execute/3.

This module is the normative home for the event names; the architecture doc (§3.7) is the source. A host attaches :telemetry handlers to these names — ALLM.Pipeline.Metrics.attach_step_handler/0 attaches the one built-in handler (populates step_logs.queue_time_ms).

Events

EventMeasurementsMetadata
[:allm_pipeline, :run, :start]%{}%{name, run_id, trigger}
[:allm_pipeline, :run, :stop]%{duration}%{name, run_id, trigger, status}
[:allm_pipeline, :run, :exception]%{duration}%{name, run_id, kind, reason}
[:allm_pipeline, :step, :start]%{}%{step_type, run_id, step_id}
[:allm_pipeline, :step, :stop]%{duration, queue_time}%{step_type, status, run_id, step_id}
[:allm_pipeline, :step, :exception]%{duration, queue_time}%{step_type, run_id, step_id, kind, reason}
[:allm_pipeline, :artifact, :store]%{bytes, compressed_bytes}%{adapter, outcome}

Units

duration and queue_time are in native time units (the :telemetry convention — System.monotonic_time/0). A consumer converts with System.convert_time_unit(value, :native, :millisecond); the built-in queue_time_ms handler does exactly that. queue_time is nil on a plain (non-fan_out) step, which has no queue.

queue_time is the backlog wait, not started_at - inserted_at

ALLM.Pipeline.StepLog.log_start/4 inserts the row with status: :running and started_at: now in the same op, so started_at - inserted_at is structurally ~0 for every real step — there is no :pending:running step-log lifecycle. queue_time is instead measured from the fan-out stage's dispatch timestamp (captured once in ALLM.Pipeline.Dsl.Runtime before Task.async_stream schedules items) threaded into each item's Context: queue_time = worker_start - fan_out_dispatch. It is non-zero for every item after the first, even at concurrency: 1 (item K waits for items 1..K-1). A plain single stage has no queue, so its queue_time is nil and that is honest, not vacuous — the metric is meaningful only on fan-outs.

Consumer accounting (earn-its-keep — ALLM_PIPELINE_DSL.md §6)

  • [:allm_pipeline, :step, :stop] has a named consumer: ALLM.Pipeline.Metrics.attach_step_handler/0, which reads queue_time and writes step_logs.queue_time_ms.
  • The :run, :step, :start, :step, :exception and :artifact families ship as a deliberately consumer-less public integration surface. That is defensible for a telemetry contract — its whole value is that a host can attach later without a framework change — but it is stated here rather than silently exempted from the earn-its-keep rule the DSL invokes everywhere else. Run stop/exception fire at the canonical settle point (ALLM.Pipeline.Lifecycle.settle/4, the DSL + owned_run/4 path); a hand-written entry point that settles through Executor.finish_run/2 instead emits :run, :start only. Reopening: a consumer that needs full run coverage routes the remaining tails through the settle point.

[:allm_pipeline, :llm, :call] is DROPPED from this phase

LLM telemetry already flows one layer down, at the ALLM-hex layer: [:allm, :generate, :stop] / [:allm, :chat, :stop], to which a host's own LLM telemetry handler is already attached (per-call model / tokens / duration). The per-step token aggregate is already persisted on step_logs.llm_total_tokens / llm_call_count and surfaced in the review UI. No consumer needs a per-step LLM telemetry event distinct from those two existing surfaces, so emitting one would be a consumer-less duplicate of the summed per-call events. Dropped per earn-its-keep. Reopening trigger: a live per-step LLM aggregate stream is needed distinct from the persisted columns.

Summary

Functions

Emit [:allm_pipeline, :artifact, :store].

Emit [:allm_pipeline, :run, :exception].

Emit [:allm_pipeline, :run, :start].

Emit [:allm_pipeline, :run, :stop].

Emit [:allm_pipeline, :step, :exception].

Emit [:allm_pipeline, :step, :start].

Emit [:allm_pipeline, :step, :stop].

The [:allm_pipeline, :step, :stop] event name (the queue_time handler attaches to it).

Functions

artifact_store(measurements, metadata)

@spec artifact_store(map(), map()) :: :ok

Emit [:allm_pipeline, :artifact, :store].

run_exception(measurements, metadata)

@spec run_exception(map(), map()) :: :ok

Emit [:allm_pipeline, :run, :exception].

run_start(metadata)

@spec run_start(map()) :: :ok

Emit [:allm_pipeline, :run, :start].

run_stop(measurements, metadata)

@spec run_stop(map(), map()) :: :ok

Emit [:allm_pipeline, :run, :stop].

step_exception(measurements, metadata)

@spec step_exception(map(), map()) :: :ok

Emit [:allm_pipeline, :step, :exception].

step_start(metadata)

@spec step_start(map()) :: :ok

Emit [:allm_pipeline, :step, :start].

step_stop(measurements, metadata)

@spec step_stop(map(), map()) :: :ok

Emit [:allm_pipeline, :step, :stop].

step_stop_event()

@spec step_stop_event() :: [atom(), ...]

The [:allm_pipeline, :step, :stop] event name (the queue_time handler attaches to it).