Current section

Files

Jump to
snakepit lib snakepit telemetry_metrics.ex
Raw

lib/snakepit/telemetry_metrics.ex

defmodule Snakepit.TelemetryMetrics do
@moduledoc """
Telemetry metric definitions and reporters for Snakepit.
Metrics focus on heartbeat and worker lifecycle events. Reporters are opt-in
via configuration under `:snakepit, :telemetry_metrics`.
"""
import Telemetry.Metrics
@type reporter_child_spec :: Supervisor.child_spec()
@default_config %{
prometheus: %{
enabled: false,
port: 9568,
name: :snakepit_prometheus_metrics,
protocol: :http
}
}
@doc """
Returns the metric definitions for Snakepit telemetry.
"""
@spec metrics() :: [Telemetry.Metrics.t()]
def metrics do
[
counter("snakepit.heartbeat.pings",
event_name: [:snakepit, :heartbeat, :ping_sent],
measurement: :pings,
tags: [:worker_id]
),
counter("snakepit.heartbeat.pongs",
event_name: [:snakepit, :heartbeat, :pong_received],
measurement: :pongs,
tags: [:worker_id]
),
counter("snakepit.heartbeat.failures",
event_name: [:snakepit, :heartbeat, :monitor_failure],
measurement: :failures,
tags: [:worker_id, :failure_reason]
),
summary("snakepit.heartbeat.latency",
event_name: [:snakepit, :heartbeat, :pong_received],
measurement: :latency_ms,
unit: :millisecond,
tags: [:worker_id]
),
last_value("snakepit.heartbeat.missed",
event_name: [:snakepit, :heartbeat, :heartbeat_timeout],
measurement: :missed,
tags: [:worker_id]
),
counter("snakepit.grpc.worker.executions",
event_name: [:snakepit, :grpc_worker, :execute, :stop],
measurement: :executions,
tags: [:worker_id, :command]
),
counter("snakepit.grpc.worker.errors",
event_name: [:snakepit, :grpc_worker, :execute, :stop],
measurement: :errors,
tags: [:worker_id, :command, :error]
),
summary("snakepit.grpc.worker.duration",
event_name: [:snakepit, :grpc_worker, :execute, :stop],
measurement: :duration_ms,
unit: :millisecond,
tags: [:worker_id, :command]
),
counter("snakepit.worker.recycled",
event_name: [:snakepit, :worker, :recycled],
measurement: :count,
tags: [:pool, :reason]
),
summary("snakepit.worker.memory_mb",
event_name: [:snakepit, :worker, :recycled],
measurement: &Map.get(&1, :memory_mb),
unit: :megabyte,
tags: [:pool]
)
]
end
@doc """
Returns reporter child specs enabled via configuration.
"""
@spec reporter_children() :: [reporter_child_spec()]
def reporter_children do
Enum.flat_map([prometheus_child_spec()], fn
nil -> []
spec -> [spec]
end)
end
defp prometheus_child_spec do
%{prometheus: prometheus_config} = load_config()
if truthy?(prometheus_config[:enabled]) do
TelemetryMetricsPrometheus.child_spec(
Keyword.merge(
[
metrics: metrics(),
port: fetch_integer(prometheus_config[:port], 9568),
name: prometheus_config[:name] || :snakepit_prometheus_metrics,
protocol: prometheus_config[:protocol] || :http
],
prometheus_extra_options(prometheus_config)
)
)
end
end
defp prometheus_extra_options(config) do
opts = []
opts =
case config[:ip] do
nil -> opts
ip when is_tuple(ip) -> Keyword.put(opts, :plug_cowboy_opts, ip: ip)
ip when is_binary(ip) -> Keyword.put(opts, :plug_cowboy_opts, ip: parse_ip(ip))
_ -> opts
end
opts
end
defp parse_ip(ip_string) do
case :inet.parse_address(String.to_charlist(ip_string)) do
{:ok, tuple} -> tuple
{:error, _reason} -> {0, 0, 0, 0}
end
end
defp truthy?(value) when is_boolean(value), do: value
defp truthy?(value) when is_binary(value) do
normalized = String.downcase(String.trim(value))
normalized in ["true", "1", "yes", "on"]
end
defp truthy?(value) when is_integer(value), do: value != 0
defp truthy?(true), do: true
defp truthy?(_), do: false
defp fetch_integer(nil, default), do: default
defp fetch_integer(value, default) when is_binary(value) do
case Integer.parse(value) do
{int, _rest} -> int
:error -> default
end
end
defp fetch_integer(value, _default) when is_integer(value), do: value
defp fetch_integer(_, default), do: default
defp load_config do
base_config =
Application.get_env(:snakepit, :telemetry_metrics, %{})
|> to_map()
Map.merge(@default_config, base_config, fn
_key, default_value, user_value when is_map(default_value) and is_map(user_value) ->
Map.merge(default_value, user_value)
_key, _default_value, user_value ->
user_value
end)
end
defp to_map(value) when is_map(value), do: value
defp to_map(value) when is_list(value) do
Enum.into(value, %{}, fn
{key, val} when is_atom(key) -> {key, to_map(val)}
{key, val} -> {String.to_atom(to_string(key)), to_map(val)}
end)
end
defp to_map(other), do: other
end