defmodule Bloccs.Web.Telemetry.Metrics do @moduledoc """ The pure functional core of the metrics collector: folds normalized bloccs telemetry into per-node rolling windows and renders a snapshot. No processes, no clock of its own — `now` (monotonic ms) is always passed in, so the whole thing is deterministically testable with synthetic events. Normalized events (produced by `Bloccs.Web.Telemetry.Handler`): * `{:start, node}` — a message entered the node * `{:stop, node, duration_ms, :ok | :failed}` — it finished * `{:exception, node}` — it raised * `{:event, node, kind}` — retry / skipped / dropped / dispatch_error A node's `state` (`:idle | :running | :ok | :failed`) drives the topology glyph; the windowed stats drive the metrics panel. """ @window_ms 10_000 @max_samples 500 @type node_state :: %{ completed: non_neg_integer(), errors: non_neg_integer(), events: non_neg_integer(), samples: [{integer(), number()}], state: :idle | :running | :ok | :failed, last_at: integer() | nil } @type t :: %{optional(atom()) => node_state()} @type event :: {:start, atom()} | {:stop, atom(), number(), :ok | :failed} | {:exception, atom()} | {:event, atom(), atom()} @type node_view :: %{ state: atom(), completed: non_neg_integer(), errors: non_neg_integer(), error_rate: float(), throughput: float(), p50: number() | nil, p95: number() | nil } @spec new() :: t() def new, do: %{} @spec apply(t(), event(), integer()) :: t() def apply(nodes, {:start, node}, now) do update(nodes, node, now, &%{&1 | state: :running}) end def apply(nodes, {:stop, node, duration_ms, outcome}, now) do update(nodes, node, now, fn m -> %{ m | completed: m.completed + 1, errors: m.errors + if(outcome == :failed, do: 1, else: 0), state: if(outcome == :failed, do: :failed, else: :ok), samples: add_sample(m.samples, {now, duration_ms}) } end) end def apply(nodes, {:exception, node}, now) do update(nodes, node, now, fn m -> %{m | completed: m.completed + 1, errors: m.errors + 1, state: :failed} end) end def apply(nodes, {:event, node, _kind}, now) do update(nodes, node, now, &%{&1 | events: &1.events + 1}) end def apply(nodes, _unknown, _now), do: nodes @doc "Render a `{node => view}` snapshot, pruning samples older than the window." @spec snapshot(t(), integer()) :: %{nodes: %{atom() => node_view()}, updated_at: integer()} def snapshot(nodes, now) do views = Map.new(nodes, fn {id, m} -> recent = prune(m.samples, now) durations = Enum.map(recent, &elem(&1, 1)) {id, %{ state: decayed_state(m, now), completed: m.completed, errors: m.errors, error_rate: ratio(m.errors, m.completed), throughput: length(recent) / (@window_ms / 1000), p50: percentile(durations, 0.5), p95: percentile(durations, 0.95) }} end) %{nodes: views, updated_at: now} end # ---- internals ---- defp blank, do: %{completed: 0, errors: 0, events: 0, samples: [], state: :idle, last_at: nil} defp update(nodes, node, now, fun) do m = Map.get(nodes, node, blank()) Map.put(nodes, node, fun.(%{m | last_at: now})) end defp add_sample(samples, sample) do [sample | samples] |> Enum.take(@max_samples) end defp prune(samples, now) do Enum.filter(samples, fn {t, _} -> now - t <= @window_ms end) end # A node that finished a while ago settles back to :idle so the graph doesn't # stay lit forever; a failure stays visible for the window. defp decayed_state(%{state: :failed} = m, now) do if stale?(m, now), do: :idle, else: :failed end defp decayed_state(%{state: :ok} = m, now) do if stale?(m, now), do: :idle, else: :ok end defp decayed_state(%{state: state}, _now), do: state defp stale?(%{last_at: nil}, _now), do: true defp stale?(%{last_at: at}, now), do: now - at > @window_ms defp ratio(_n, 0), do: 0.0 defp ratio(n, d), do: n / d defp percentile([], _q), do: nil defp percentile(values, q) do sorted = Enum.sort(values) idx = max(0, round(q * (length(sorted) - 1))) Enum.at(sorted, idx) end end