defmodule PromEx.Plugins.Beam do
@moduledoc """
Telemetry metrics for the BEAM.
This plugin captures metrics regarding the Erlang Virtual Machine (i.e the BEAM). Specifically, it captures metrics
regarding the CPU topology, system limits, VM feature support, scheduler information, memory utilization, distribution
traffic, and other internal metrics.
This plugin supports the following options:
- `poll_rate`: This is option is OPTIONAL and is the rate at which poll metrics are refreshed (default is 5 seconds).
- `metric_prefix`: This option is OPTIONAL and is used to override the default metric prefix of
`[otp_app, :prom_ex, :beam]`. If this changes you will also want to set `beam_metric_prefix`
in your `dashboard_assigns` to the snakecase version of your prefix, the default
`beam_metric_prefix` is `{otp_app}_prom_ex_beam`.
- `duration_unit`: This is an OPTIONAL option and is a `Telemetry.Metrics.time_unit()`. It can be one of:
`:second | :millisecond | :microsecond | :nanosecond`. It is `:millisecond` by default.
This plugin exposes the following metric groups:
- `:beam_memory_polling_metrics`
- `:beam_internal_polling_metrics`
- `:beam_cpu_topology_manual_metrics`
- `:beam_system_limits_manual_metrics`
- `:beam_system_info_manual_metrics`
- `:beam_scheduler_manual_metrics`
To use plugin in your application, add the following to your PromEx module:
```
defmodule MyApp.PromEx do
use PromEx, otp_app: :web_app
@impl true
def plugins do
[
...
PromEx.Plugins.Beam
]
end
@impl true
def dashboards do
[
...
{:prom_ex, "beam.json"}
]
end
end
```
This plugin exposes manual metrics so be sure to configure the PromEx `:delay_manual_start` as necessary.
"""
use PromEx.Plugin
@memory_event [:prom_ex, :plugin, :beam, :memory]
@impl true
def polling_metrics(opts) do
poll_rate = Keyword.get(opts, :poll_rate, 5_000)
otp_app = Keyword.fetch!(opts, :otp_app)
metric_prefix = Keyword.get(opts, :metric_prefix, PromEx.metric_prefix(otp_app, :beam))
duration_unit = Keyword.get(opts, :duration_unit, :millisecond)
# TODO: Investigate Microstate accounting metrics
# http://erlang.org/doc/man/erlang.html#statistics_microstate_accounting
# TODO: Add a metrics group for allocators
# https://erlang.org/doc/man/erts_alloc.html
# :erlang.system_info(:allocator)
[
memory_metrics(metric_prefix, poll_rate),
mnesia_metrics(metric_prefix, poll_rate),
distribution_metrics(metric_prefix, poll_rate),
beam_internal_metrics(metric_prefix, poll_rate, duration_unit)
]
end
@impl true
def manual_metrics(opts) do
otp_app = Keyword.fetch!(opts, :otp_app)
metric_prefix = Keyword.get(opts, :metric_prefix, PromEx.metric_prefix(otp_app, :beam))
[
beam_cpu_topology_info(metric_prefix),
beam_system_limits_info(metric_prefix),
beam_system_info(metric_prefix),
beam_scheduler_info(metric_prefix)
]
end
defp distribution_metrics(_metric_prefix, poll_rate) do
Polling.build(
:beam_distribution_polling_metrics,
poll_rate,
{__MODULE__, :execute_distribution_metrics, []},
[]
)
end
defp mnesia_metrics(_metric_prefix, poll_rate) do
Polling.build(
:beam_mnesia_polling_metrics,
poll_rate,
{__MODULE__, :execute_mnesia_metrics, []},
[]
)
end
defp beam_internal_metrics(metric_prefix, poll_rate, duration_unit) do
duration_unit_plural = String.to_atom("#{duration_unit}s")
Polling.build(
:beam_internal_polling_metrics,
poll_rate,
{__MODULE__, :execute_internal_metrics, []},
[
last_value(
metric_prefix ++ [:stats, :active_task, :count],
event_name: [:prom_ex, :plugin, :beam, :active_task, :count],
description: "The number of processes and ports that are ready to run, or are currently running.",
measurement: :count,
tags: [:type]
),
last_value(
metric_prefix ++ [:stats, :run_queue, :count],
event_name: [:prom_ex, :plugin, :beam, :run_queue, :count],
description: "The number of processes and ports that are ready to run and are in the run queue.",
measurement: :count,
tags: [:type]
),
last_value(
metric_prefix ++ [:stats, :context_switch, :count],
event_name: [:prom_ex, :plugin, :beam, :context_switch, :count],
description: "The total number of context switches since the system started.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :reduction, :count],
event_name: [:prom_ex, :plugin, :beam, :reduction, :count],
description: "The total number of reductions since the system started.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :gc, :count],
event_name: [:prom_ex, :plugin, :beam, :gc, :count],
description: "The total number of garbage collections since the system started.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :gc, :reclaimed, :bytes],
event_name: [:prom_ex, :plugin, :beam, :gc, :bytes_reclaimed],
description: "The total number of bytes reclaimed since the system started.",
measurement: :count,
unit: :byte
),
last_value(
metric_prefix ++ [:stats, :port_io, :byte, :count],
event_name: [:prom_ex, :plugin, :beam, :port_io, :count],
description: "The total number of bytes sent and received through ports since the system started.",
measurement: :count,
tags: [:type],
unit: :byte
),
last_value(
metric_prefix ++ [:stats, :uptime, duration_unit_plural, :count],
event_name: [:prom_ex, :plugin, :beam, :uptime, :count],
description:
"The total number of wall clock #{duration_unit_plural} that have passed since the system started.",
measurement: :count,
unit: duration_unit
),
last_value(
metric_prefix ++ [:stats, :port, :count],
event_name: [:prom_ex, :plugin, :beam, :port, :count],
description: "A count of how many ports are currently active.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :process, :count],
event_name: [:prom_ex, :plugin, :beam, :process, :count],
description: "A count of how many Erlang processes are currently running.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :atom, :count],
event_name: [:prom_ex, :plugin, :beam, :atom, :count],
description: "A count of how many atoms are currently allocated.",
measurement: :count
),
last_value(
metric_prefix ++ [:stats, :ets, :count],
event_name: [:prom_ex, :plugin, :beam, :ets, :count],
description: "A count of how many ETS tables currently exist.",
measurement: :count
)
]
)
end
defp beam_system_info(metric_prefix) do
Manual.build(
:beam_system_info_manual_metrics,
{__MODULE__, :execute_system_info, []},
[
last_value(
metric_prefix ++ [:system, :version, :info],
event_name: [:prom_ex, :plugin, :beam, :version],
description: "The OTP release major version.",
measurement: :version
),
last_value(
metric_prefix ++ [:system, :smp_support, :info],
event_name: [:prom_ex, :plugin, :beam, :smp_support],
description: "Whether the BEAM instance has been compiled with SMP support.",
measurement: :enabled
),
last_value(
metric_prefix ++ [:system, :jit_support, :info],
event_name: [:prom_ex, :plugin, :beam, :jit_support],
description: "Whether the BEAM instance is running with the JIT compiler.",
measurement: :enabled
),
last_value(
metric_prefix ++ [:system, :thread_support, :info],
event_name: [:prom_ex, :plugin, :beam, :thread_support],
description: "Whether the BEAM instance has been compiled with threading support.",
measurement: :enabled
),
last_value(
metric_prefix ++ [:system, :time_correction_support, :info],
event_name: [:prom_ex, :plugin, :beam, :time_correction_support],
description: "Whether the BEAM instance has time correction support.",
measurement: :enabled
),
last_value(
metric_prefix ++ [:system, :word_size_bytes, :info],
event_name: [:prom_ex, :plugin, :beam, :word_size_bytes],
description: "The size of Erlang term words in bytes.",
measurement: :size
)
]
)
end
defp beam_scheduler_info(metric_prefix) do
Manual.build(
:beam_scheduler_manual_metrics,
{__MODULE__, :execute_scheduler_info, []},
[
last_value(
metric_prefix ++ [:system, :dirty_cpu_schedulers, :info],
event_name: [:prom_ex, :plugin, :beam, :dirty_cpu_schedulers],
description: "The total number of dirty CPU scheduler threads used by the BEAM.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :dirty_cpu_schedulers_online, :info],
event_name: [:prom_ex, :plugin, :beam, :dirty_cpu_schedulers_online],
description: "The total number of dirty CPU schedulers that are online.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :dirty_io_schedulers, :info],
event_name: [:prom_ex, :plugin, :beam, :dirty_io_schedulers],
description: "The total number of dirty I/O schedulers used to execute I/O bound native functions.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :schedulers, :info],
event_name: [:prom_ex, :plugin, :beam, :schedulers],
description: "The number of scheduler threads in use by the BEAM.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :schedulers_online, :info],
event_name: [:prom_ex, :plugin, :beam, :schedulers_online],
description: "The number of scheduler threads that are online.",
measurement: :quantity
)
]
)
end
defp beam_cpu_topology_info(metric_prefix) do
Manual.build(
:beam_cpu_topology_manual_metrics,
{__MODULE__, :execute_cpu_topology_info, []},
[
last_value(
metric_prefix ++ [:system, :logical_processors, :info],
event_name: [:prom_ex, :plugin, :beam, :logical_processors],
description: "The total number of logical processors on the host machine.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :logical_processors_available, :info],
event_name: [:prom_ex, :plugin, :beam, :logical_processors_available],
description: "The total number of logical processors available to the BEAM.",
measurement: :quantity
),
last_value(
metric_prefix ++ [:system, :logical_processors_online, :info],
event_name: [:prom_ex, :plugin, :beam, :logical_processors_online],
description: "The total number of logical processors online on the host machine.",
measurement: :quantity
)
]
)
end
defp beam_system_limits_info(metric_prefix) do
Manual.build(
:beam_system_limits_manual_metrics,
{__MODULE__, :execute_system_limits_info, []},
[
last_value(
metric_prefix ++ [:system, :ets_limit, :info],
event_name: [:prom_ex, :plugin, :beam, :ets_limit],
description:
"The maximum number of ETS tables allowed (this is partially obsolete given that the number of ETS tables is limited by available memory).",
measurement: :limit
),
last_value(
metric_prefix ++ [:system, :port_limit, :info],
event_name: [:prom_ex, :plugin, :beam, :port_limit],
description: "The maximum number of ports that can simultaneously exist on the BEAM instance.",
measurement: :limit
),
last_value(
metric_prefix ++ [:system, :process_limit, :info],
event_name: [:prom_ex, :plugin, :beam, :process_limit],
description: "The maximum number of processes that can simultaneously exist on the BEAM instance.",
measurement: :limit
),
last_value(
metric_prefix ++ [:system, :thread_pool_size, :info],
event_name: [:prom_ex, :plugin, :beam, :thread_pool_size],
description: "The number of async threads in the async threads pool used for async driver calls.",
measurement: :size
),
last_value(
metric_prefix ++ [:system, :atom_limit, :info],
event_name: [:prom_ex, :plugin, :beam, :atom_limit],
description: "The maximum number of atoms allowed.",
measurement: :limit
)
]
)
end
defp memory_metrics(metric_prefix, poll_rate) do
Polling.build(
:beam_memory_polling_metrics,
poll_rate,
{__MODULE__, :execute_memory_metrics, []},
[
# Capture the total memory allocated to the entire Erlang VM (or BEAM for short)
last_value(
metric_prefix ++ [:memory, :allocated, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated.",
measurement: :total,
unit: :byte
),
# Capture the total memory allocated to atoms
last_value(
metric_prefix ++ [:memory, :atom, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated for atoms.",
measurement: :atom,
unit: :byte
),
# Capture the total memory allocated to binaries
last_value(
metric_prefix ++ [:memory, :binary, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated for binaries.",
measurement: :binary,
unit: :byte
),
# Capture the total memory allocated to Erlang code
last_value(
metric_prefix ++ [:memory, :code, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated for Erlang code.",
measurement: :code,
unit: :byte
),
# Capture the total memory allocated to ETS tables
last_value(
metric_prefix ++ [:memory, :ets, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated for ETS tables.",
measurement: :ets,
unit: :byte
),
# Capture the total memory allocated to Erlang processes
last_value(
metric_prefix ++ [:memory, :processes, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated to Erlang processes.",
measurement: :processes,
unit: :byte
),
# Capture the total memory allocated to :persistent_term
last_value(
metric_prefix ++ [:memory, :persistent_term, :total, :bytes],
event_name: @memory_event,
description: "The total amount of memory currently allocated to Erlang :persistent_term.",
measurement: :persistent_term,
unit: :byte
)
]
)
end
@doc false
def execute_memory_metrics do
%{memory: persistent_term_memory} = :persistent_term.info()
memory_measurements =
:erlang.memory()
|> Map.new()
|> Map.put(:persistent_term, persistent_term_memory)
:telemetry.execute(@memory_event, memory_measurements, %{})
end
@doc false
def execute_distribution_metrics do
end
@doc false
def execute_internal_metrics do
total_active_tasks = :erlang.statistics(:total_active_tasks)
total_active_tasks_all = :erlang.statistics(:total_active_tasks_all)
total_run_queue_lengths = :erlang.statistics(:total_run_queue_lengths)
total_run_queue_lengths_all = :erlang.statistics(:total_run_queue_lengths_all)
dirty_active_tasks = total_active_tasks_all - total_active_tasks
dirty_run_queue_lengths = total_run_queue_lengths_all - total_run_queue_lengths
{context_switches, _} = :erlang.statistics(:context_switches)
{total_reductions, _} = :erlang.statistics(:reductions)
word_size = :erlang.system_info(:wordsize)
{number_of_gcs, words_reclaimed, _} = :erlang.statistics(:garbage_collection)
bytes_reclaimed = words_reclaimed * word_size
{{:input, input_port_bytes}, {:output, output_port_bytes}} = :erlang.statistics(:io)
{wall_clock_time, _} = :erlang.statistics(:wall_clock)
:telemetry.execute([:prom_ex, :plugin, :beam, :port, :count], %{count: :erlang.system_info(:port_count)})
:telemetry.execute([:prom_ex, :plugin, :beam, :process, :count], %{count: :erlang.system_info(:process_count)})
:telemetry.execute([:prom_ex, :plugin, :beam, :atom, :count], %{count: :erlang.system_info(:atom_count)})
:telemetry.execute([:prom_ex, :plugin, :beam, :ets, :count], %{count: :erlang.system_info(:ets_count)})
:telemetry.execute([:prom_ex, :plugin, :beam, :active_task, :count], %{count: total_active_tasks}, %{type: :normal})
:telemetry.execute([:prom_ex, :plugin, :beam, :active_task, :count], %{count: dirty_active_tasks}, %{type: :dirty})
:telemetry.execute([:prom_ex, :plugin, :beam, :context_switch, :count], %{count: context_switches})
:telemetry.execute([:prom_ex, :plugin, :beam, :reduction, :count], %{count: total_reductions})
:telemetry.execute([:prom_ex, :plugin, :beam, :gc, :count], %{count: number_of_gcs})
:telemetry.execute([:prom_ex, :plugin, :beam, :gc, :bytes_reclaimed], %{count: bytes_reclaimed})
:telemetry.execute([:prom_ex, :plugin, :beam, :port_io, :count], %{count: input_port_bytes}, %{type: :input})
:telemetry.execute([:prom_ex, :plugin, :beam, :port_io, :count], %{count: output_port_bytes}, %{type: :output})
:telemetry.execute([:prom_ex, :plugin, :beam, :uptime, :count], %{count: wall_clock_time})
:telemetry.execute([:prom_ex, :plugin, :beam, :run_queue, :count], %{count: total_run_queue_lengths}, %{
type: :normal
})
:telemetry.execute([:prom_ex, :plugin, :beam, :run_queue, :count], %{count: dirty_run_queue_lengths}, %{
type: :dirty
})
end
@doc false
def execute_mnesia_metrics do
# https://github.com/deadtrickster/prometheus.erl/blob/master/src/collectors/mnesia/prometheus_mnesia_collector.erl
end
@doc false
def execute_system_limits_info do
:telemetry.execute([:prom_ex, :plugin, :beam, :ets_limit], %{limit: :erlang.system_info(:ets_limit)})
:telemetry.execute([:prom_ex, :plugin, :beam, :port_limit], %{limit: :erlang.system_info(:port_limit)})
:telemetry.execute([:prom_ex, :plugin, :beam, :process_limit], %{limit: :erlang.system_info(:process_limit)})
:telemetry.execute(
[:prom_ex, :plugin, :beam, :thread_pool_size],
%{size: :erlang.system_info(:thread_pool_size)},
%{}
)
:telemetry.execute([:prom_ex, :plugin, :beam, :atom_limit], %{limit: :erlang.system_info(:atom_limit)}, %{})
end
@doc false
def execute_system_info do
smp_enabled = if(:erlang.system_info(:smp_support), do: 1, else: 0)
thread_support_enabled = if(:erlang.system_info(:threads), do: 1, else: 0)
time_correction_enabled = if(:erlang.system_info(:time_correction), do: 1, else: 0)
word_size = :erlang.system_info(:wordsize)
version = :otp_release |> :erlang.system_info() |> :erlang.list_to_binary() |> String.to_integer()
jit_enabled =
try do
if :erlang.system_info(:emu_flavor) == :jit, do: 1, else: 0
rescue
_error ->
0
end
:telemetry.execute([:prom_ex, :plugin, :beam, :smp_support], %{enabled: smp_enabled}, %{})
:telemetry.execute([:prom_ex, :plugin, :beam, :jit_support], %{enabled: jit_enabled}, %{})
:telemetry.execute([:prom_ex, :plugin, :beam, :thread_support], %{enabled: thread_support_enabled}, %{})
:telemetry.execute([:prom_ex, :plugin, :beam, :time_correction_support], %{enabled: time_correction_enabled}, %{})
:telemetry.execute([:prom_ex, :plugin, :beam, :word_size_bytes], %{size: word_size}, %{})
:telemetry.execute([:prom_ex, :plugin, :beam, :version], %{version: version}, %{})
end
@doc false
def execute_cpu_topology_info do
:telemetry.execute(
[:prom_ex, :plugin, :beam, :logical_processors],
%{quantity: :erlang.system_info(:logical_processors)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :logical_processors_available],
%{quantity: :erlang.system_info(:logical_processors_available)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :logical_processors_online],
%{quantity: :erlang.system_info(:logical_processors_online)},
%{}
)
end
@doc false
def execute_scheduler_info do
:telemetry.execute(
[:prom_ex, :plugin, :beam, :dirty_cpu_schedulers],
%{quantity: :erlang.system_info(:dirty_cpu_schedulers)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :dirty_cpu_schedulers_online],
%{quantity: :erlang.system_info(:dirty_cpu_schedulers_online)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :dirty_io_schedulers],
%{quantity: :erlang.system_info(:dirty_io_schedulers)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :schedulers],
%{quantity: :erlang.system_info(:schedulers)},
%{}
)
:telemetry.execute(
[:prom_ex, :plugin, :beam, :schedulers_online],
%{quantity: :erlang.system_info(:schedulers_online)},
%{}
)
end
end
