ObserverWeb.Processes (Observer Web v0.2.8)

Samples and ranks the processes of a node, etop-style.

Sampling delegates to :observer_backend.etop_collect (part of runtime_tools, which Observer Web already requires on every node for :dbg tracing) - the same collector etop itself uses against remote nodes, so one RPC round trip returns every process's memory, cumulative reductions, message queue length, name and current function, regardless of the observer_web version (or absence) on the observed node.

NOTE: :observer_backend.etop_collect has a documented side effect: when the :scheduler_wall_time system flag is off, it turns the flag on and holds it (via a monitor) until the collector process - here, the LiveView showing the Processes page - dies. This is exactly how running etop against a node behaves; the flag adds a small scheduler accounting cost and is switched back off when the page is closed.

Reductions are cumulative counters, so ranking by "load" uses the delta between two consecutive samples (rank/2 with the previous sample's reductions map) - the first sample ranks by the cumulative value, exactly like etop's first tick.

Summary

Types

process_row()

sample()

Functions

details(pid)

Fetches display-ready details for one process, used by the Processes page drill-down panel. Returns {:error, :not_found} when the process has already died.

rank(map, sort_by, limit, previous_reductions \\ %{})

Ranks a sample's processes by sort_by, keeping the top limit rows.

reductions_by_pid(map)

Extracts the %{pid => cumulative_reductions} map used as previous_reductions for the next rank/4 call.

sample(node \\ Node.self(), timeout \\ 10000)

Collects one sample of every process on node.

Types

process_row()

@type process_row() :: %{
  pid: pid(),
  name: String.t(),
  memory: non_neg_integer(),
  reductions: non_neg_integer(),
  reductions_diff: non_neg_integer(),
  message_queue_len: non_neg_integer(),
  current_function: String.t()
}

sample()

@type sample() :: %{
  node: node(),
  process_count: non_neg_integer(),
  run_queue: non_neg_integer(),
  memory: %{optional(atom()) => non_neg_integer()},
  processes: [process_row()]
}

Functions

details(pid)

@spec details(pid()) :: {:ok, [{String.t(), String.t()}]} | {:error, :not_found}

Fetches display-ready details for one process, used by the Processes page drill-down panel. Returns {:error, :not_found} when the process has already died.

rank(map, sort_by, limit, previous_reductions \\ %{})

@spec rank(sample(), :reductions | :memory | :message_queue_len, pos_integer(), map()) ::
  [
    process_row()
  ]

Ranks a sample's processes by sort_by, keeping the top limit rows.

previous_reductions is a %{pid => cumulative_reductions} map from the preceding sample; each row's reductions_diff becomes the delta since then (processes not present before - or on the very first sample - keep their cumulative count, like etop's first tick).

reductions_by_pid(map)

@spec reductions_by_pid(sample()) :: %{optional(pid()) => non_neg_integer()}

Extracts the %{pid => cumulative_reductions} map used as previous_reductions for the next rank/4 call.

sample(node \\ Node.self(), timeout \\ 10000)

@spec sample(node(), timeout()) :: {:ok, sample()} | {:error, term()}

Collects one sample of every process on node.

The #etop_info{}/#etop_proc_info{} records are decoded positionally - their shape is part of runtime_tools' cross-node collector contract and has been stable across OTP releases (the GUI observer and etop rely on it the same way). Anything unexpected is reported as an error instead of crashing the caller.