Export BEAM processes to OTLP metrics endpoint (plausible#6188)

* Export BEAM processes to OTLP metrics endpoint

* fix max nesting

* update mix.lock

* switch to Enum.map insted

Author: cnkk

config/runtime.exs

@@ -960,6 +960,30 @@ else
     traces_exporter: :none
 end
 
+beam_metrics_enabled? = get_bool_from_path_or_env(config_dir, "BEAM_METRICS_ENABLED", false)
+
+if beam_metrics_enabled? do
+  beam_metrics_interval = get_int_from_path_or_env(config_dir, "BEAM_METRICS_INTERVAL_MS", 5_000)
+
+  beam_metrics_otlp_endpoint =
+    get_var_from_path_or_env(config_dir, "OTEL_EXPORTER_OTLP_ENDPOINT") || otlp_endpoint
+
+  config :opentelemetry_experimental,
+    readers: [
+      %{
+        module: :otel_metric_reader,
+        config: %{
+          export_interval_ms: beam_metrics_interval,
+          exporter:
+            {:otel_exporter_metrics_otlp,
+             %{
+               endpoints: [beam_metrics_otlp_endpoint]
+             }}
+        }
+      }
+    ]
+end
+
 config :tzdata, :data_dir, Path.join(persistent_cache_dir || System.tmp_dir!(), "tzdata_data")
 
 promex_disabled? = get_bool_from_path_or_env(config_dir, "PROMEX_DISABLED", true)

lib/plausible/application.ex

@@ -373,6 +373,10 @@ defmodule Plausible.Application do
     OpentelemetryEcto.setup([:plausible, :clickhouse_repo], db_statement: :enabled)
     OpentelemetryOban.setup()
     Plausible.OpenTelemetry.Logger.setup()
+
+    if Application.get_env(:opentelemetry_experimental, :readers, []) != [] do
+      Plausible.OpenTelemetry.BeamMetrics.setup()
+    end
   end
 
   defp setup_geolocation do

lib/plausible/open_telemetry/beam_metrics.ex

@@ -0,0 +1,107 @@
+defmodule Plausible.OpenTelemetry.BeamMetrics do
+  @moduledoc """
+  Periodic BEAM process sampling via OpenTelemetry observable gauges.
+
+  Uses `:recon.proc_count/2` to sample top-N processes by memory, reductions,
+  and message queue length. Emits data as OTel observable gauge observations,
+  exported via OTLP to the configured OTel Collector.
+
+  Disabled by default. Enable with `BEAM_METRICS_ENABLED=true`.
+  Collection interval controlled by `BEAM_METRICS_INTERVAL_MS` (default: 5000).
+  """
+
+  require Logger
+
+  @top_n 20
+  @metrics [:memory, :reductions, :message_queue_len]
+  @process_info_keys [
+    :registered_name,
+    :current_function,
+    :initial_call,
+    :memory,
+    :reductions,
+    :message_queue_len
+  ]
+
+  @instruments %{
+    memory:
+      {:"beam.top_process.memory",
+       %{description: "Memory usage of top BEAM processes", unit: :bytes}},
+    reductions:
+      {:"beam.top_process.reductions",
+       %{description: "Reductions of top BEAM processes", unit: :"1"}},
+    message_queue_len:
+      {:"beam.top_process.message_queue_len",
+       %{description: "Message queue length of top BEAM processes", unit: :"1"}}
+  }
+
+  @doc """
+  Registers OTel observable gauge instruments and a shared callback.
+
+  Should be called once during application startup when BEAM metrics are enabled.
+  """
+  def setup do
+    scope = :opentelemetry.instrumentation_scope("plausible_beam_metrics", "0.1.0", :undefined)
+    meter = :opentelemetry_experimental.get_meter(scope)
+
+    gauges =
+      Enum.map(@metrics, fn metric ->
+        {name, opts} = Map.fetch!(@instruments, metric)
+        :otel_meter.create_observable_gauge(meter, name, opts)
+      end)
+
+    :otel_meter.register_callback(meter, gauges, &observe_top_processes/1, [])
+
+    Logger.info("BEAM metrics setup complete — sampling top #{@top_n} processes per metric")
+    :ok
+  end
+
+  @doc """
+  Callback invoked by the OTel Metric Reader on each collection cycle.
+
+  Returns named observations for all three gauge instruments.
+  """
+  def observe_top_processes(_callback_args) do
+    Enum.map(@metrics, fn metric ->
+      {gauge_name, _opts} = Map.fetch!(@instruments, metric)
+      observations = collect_observations(metric)
+      {gauge_name, observations}
+    end)
+  end
+
+  defp collect_observations(metric) do
+    metric
+    |> :recon.proc_count(@top_n)
+    |> Enum.flat_map(fn {pid, value, _info} ->
+      case Process.info(pid, @process_info_keys) do
+        nil -> []
+        info -> [{value, build_attributes(pid, info)}]
+      end
+    end)
+  end
+
+  defp build_attributes(pid, info) do
+    registered_name =
+      case Keyword.get(info, :registered_name) do
+        [] -> ""
+        name when is_atom(name) -> Atom.to_string(name)
+        _ -> ""
+      end
+
+    current_function = format_mfa(Keyword.get(info, :current_function))
+    initial_call = format_mfa(Keyword.get(info, :initial_call))
+
+    %{
+      "beam.process.pid" => inspect(pid),
+      "beam.process.registered_name" => registered_name,
+      "beam.process.current_function" => current_function,
+      "beam.process.initial_call" => initial_call,
+      "beam.process.memory" => to_string(Keyword.get(info, :memory, 0)),
+      "beam.process.reductions" => to_string(Keyword.get(info, :reductions, 0)),
+      "beam.process.message_queue_len" => to_string(Keyword.get(info, :message_queue_len, 0))
+    }
+  end
+
+  defp format_mfa({m, f, a}), do: Exception.format_mfa(m, f, a)
+  defp format_mfa(_), do: ""
+end

mix.exs

@@ -109,8 +109,18 @@ defmodule Plausible.MixProject do
       {:observer_cli, "~> 1.7"},
       {:opentelemetry, "~> 1.7"},
       {:opentelemetry_api, "~> 1.5"},
+      {:opentelemetry_api_experimental,
+       git: "https://github.com/open-telemetry/opentelemetry-erlang.git",
+       ref: "b241ebf4feb7558738d009583bae8602b3c49fd0",
+       sparse: "apps/opentelemetry_api_experimental",
+       override: true},
       {:opentelemetry_ecto, "~> 1.2"},
       {:opentelemetry_exporter, "~> 1.10"},
+      {:opentelemetry_experimental,
+       git: "https://github.com/open-telemetry/opentelemetry-erlang.git",
+       ref: "b241ebf4feb7558738d009583bae8602b3c49fd0",
+       sparse: "apps/opentelemetry_experimental",
+       override: true},
       {:opentelemetry_phoenix, "~> 2.0.1"},
       {:opentelemetry_oban, "~> 1.1"},
       {:opentelemetry_cowboy, "~> 1.0"},
@@ -132,6 +142,7 @@ defmodule Plausible.MixProject do
       {:prom_ex, "~> 1.8"},
       {:peep, "~> 3.0"},
       {:public_suffix, git: "https://github.com/axelson/publicsuffix-elixir"},
+      {:recon, "~> 2.5"},
       {:ref_inspector, "~> 2.0"},
       {:referrer_blocklist, git: "https://github.com/plausible/referrer-blocklist.git"},
       {:sentry, "~> 11.0.4"},

mix.lock

@@ -106,8 +106,10 @@
   "open_api_spex": {:hex, :open_api_spex, "3.22.1", "38c99d8bf107dc7ffb112dc669f33e0a287396c6c9fd6bb7eeb27e3fe8dbba0e", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:poison, "~> 3.0 or ~> 4.0 or ~> 5.0 or ~> 6.0", [hex: :poison, repo: "hexpm", optional: true]}, {:ymlr, "~> 2.0 or ~> 3.0 or ~> 4.0 or ~> 5.0", [hex: :ymlr, repo: "hexpm", optional: true]}], "hexpm", "fa51ecd04ececbad89a8ede55ebd9db7aa9e55cc7ddbb46455522e0f3c098290"},
   "opentelemetry": {:hex, :opentelemetry, "1.7.0", "20d0f12d3d1c398d3670fd44fd1a7c495dd748ab3e5b692a7906662e2fb1a38a", [:rebar3], [{:opentelemetry_api, "~> 1.5.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "a9173b058c4549bf824cbc2f1d2fa2adc5cdedc22aa3f0f826951187bbd53131"},
   "opentelemetry_api": {:hex, :opentelemetry_api, "1.5.0", "1a676f3e3340cab81c763e939a42e11a70c22863f645aa06aafefc689b5550cf", [:mix, :rebar3], [], "hexpm", "f53ec8a1337ae4a487d43ac89da4bd3a3c99ddf576655d071deed8b56a2d5dda"},
+  "opentelemetry_api_experimental": {:git, "https://github.com/open-telemetry/opentelemetry-erlang.git", "b241ebf4feb7558738d009583bae8602b3c49fd0", [ref: "b241ebf4feb7558738d009583bae8602b3c49fd0", sparse: "apps/opentelemetry_api_experimental"]},
   "opentelemetry_cowboy": {:hex, :opentelemetry_cowboy, "1.0.0", "786c7cde66a2493323c79d2c94e679ff501d459a9b403d8b60b9bef116333117", [:rebar3], [{:cowboy_telemetry, "~> 0.4", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 1.27", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.1", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:otel_http, "~> 0.2", [hex: :otel_http, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7575716eaccacd0eddc3e7e61403aecb5d0a6397183987d6049094aeb0b87a7c"},
   "opentelemetry_ecto": {:hex, :opentelemetry_ecto, "1.2.0", "2382cb47ddc231f953d3b8263ed029d87fbf217915a1da82f49159d122b64865", [:mix], [{:opentelemetry_api, "~> 1.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.2", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "70dfa2e79932e86f209df00e36c980b17a32f82d175f0068bf7ef9a96cf080cf"},
+  "opentelemetry_experimental": {:git, "https://github.com/open-telemetry/opentelemetry-erlang.git", "b241ebf4feb7558738d009583bae8602b3c49fd0", [ref: "b241ebf4feb7558738d009583bae8602b3c49fd0", sparse: "apps/opentelemetry_experimental"]},
   "opentelemetry_exporter": {:hex, :opentelemetry_exporter, "1.10.0", "972e142392dbfa679ec959914664adefea38399e4f56ceba5c473e1cabdbad79", [:rebar3], [{:grpcbox, ">= 0.0.0", [hex: :grpcbox, repo: "hexpm", optional: false]}, {:opentelemetry, "~> 1.7.0", [hex: :opentelemetry, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.5.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:tls_certificate_check, "~> 1.18", [hex: :tls_certificate_check, repo: "hexpm", optional: false]}], "hexpm", "33a116ed7304cb91783f779dec02478f887c87988077bfd72840f760b8d4b952"},
   "opentelemetry_oban": {:hex, :opentelemetry_oban, "1.1.1", "519e9ba60d3dc3483ad2df3fade131d47056e0dae74f0724c8a40b9718f089d1", [:mix], [{:oban, "~> 2.0", [hex: :oban, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.1", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ae6aed431626a94a4bb6bf5b268247ced687ec8f99eced6887e3754f9d3a2089"},
   "opentelemetry_phoenix": {:hex, :opentelemetry_phoenix, "2.0.1", "c664cdef205738cffcd409b33599439a4ffb2035ef6e21a77927ac1da90463cb", [:mix], [{:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_process_propagator, "~> 0.3", [hex: :opentelemetry_process_propagator, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 1.27", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:opentelemetry_telemetry, "~> 1.1", [hex: :opentelemetry_telemetry, repo: "hexpm", optional: false]}, {:otel_http, "~> 0.2", [hex: :otel_http, repo: "hexpm", optional: false]}, {:plug, ">= 1.11.0", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "a24fdccdfa6b890c8892c6366beab4a15a27ec0c692b0f77ec2a862e7b235f6e"},