Packages
electric
1.2.3
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/telemetry/system_monitor.ex
defmodule Electric.Telemetry.SystemMonitor do
@moduledoc """
Application-wide process that initializes Erlang's system monitor and consumes monitoring events.
Currently the follow events are tracked:
- long_gc
- long_schedule
- long_message_queue
"""
use GenServer
import Electric.Telemetry.Processes, only: [proc_type: 1]
require Logger
@vm_monitor_long_gc [:vm, :monitor, :long_gc]
@vm_monitor_long_schedule [:vm, :monitor, :long_schedule]
@vm_monitor_long_message_queue [:vm, :monitor, :long_message_queue]
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
def init(opts) do
:erlang.system_monitor(self(),
long_gc: opts.long_gc_threshold,
long_schedule: opts.long_schedule_threshold,
long_message_queue:
{opts.long_message_queue_disable_threshold, opts.long_message_queue_enable_threshold}
)
{:ok, %{long_message_queue_pids: %{}, long_message_queue_timer: nil}}
end
def handle_info({:monitor, gc_pid, :long_gc, info}, state) do
type = proc_type(gc_pid)
Logger.debug(
"Long GC detected for pid #{inspect(gc_pid)} (#{inspect(type)}): took #{Keyword.fetch!(info, :timeout)}ms. #{inspect(info, limit: :infinity)}"
)
:telemetry.execute(@vm_monitor_long_gc, Map.new(info), %{process_type: type})
{:noreply, state}
end
def handle_info({:monitor, port, :long_schedule, info}, state) when is_port(port) do
Logger.debug(
"Long schedule detected for port #{inspect(port)}, took #{Keyword.fetch!(info, :timeout)}ms"
)
:telemetry.execute(@vm_monitor_long_schedule, Map.new(info), %{process_type: :port})
{:noreply, state}
end
def handle_info({:monitor, pid, :long_schedule, info}, state) when is_pid(pid) do
type = proc_type(pid)
Logger.debug(fn ->
locations =
info
|> Keyword.delete(:timeout)
|> Map.new(fn {loc, {m, f, a}} when loc in [:in, :out] ->
{loc, Exception.format_mfa(m, f, a)}
end)
locs_str =
if map_size(locations) > 0 do
"; " <>
(locations |> Enum.map(fn {key, val} -> "#{key}: #{val}" end) |> Enum.join(", "))
else
""
end
"Long schedule detected for pid #{inspect(pid)} (#{inspect(type)}), took #{Keyword.fetch!(info, :timeout)}ms" <>
locs_str
end)
:telemetry.execute(@vm_monitor_long_schedule, %{timeout: Keyword.fetch!(info, :timeout)}, %{
process_type: type
})
{:noreply, state}
end
def handle_info({:monitor, pid, :long_message_queue, true}, state) do
type = proc_type(pid)
Logger.debug("Long message queue detected for pid #{inspect(pid)} (#{inspect(type)})")
log_long_message_queue_event(pid, type)
state =
%{state | long_message_queue_pids: Map.put(state.long_message_queue_pids, pid, type)}
|> maybe_start_long_message_queue_timer()
{:noreply, state}
end
def handle_info({:monitor, pid, :long_message_queue, false}, state) do
Logger.debug("Long message queue no longer detected for pid #{inspect(pid)}")
{:noreply, %{state | long_message_queue_pids: Map.delete(state.long_message_queue_pids, pid)}}
end
def handle_info(:recheck_message_queues, state)
when map_size(state.long_message_queue_pids) == 0 do
:timer.cancel(state.long_message_queue_timer)
{:noreply, %{state | long_message_queue_timer: nil}}
end
def handle_info(:recheck_message_queues, state) do
Enum.each(state.long_message_queue_pids, fn {pid, type} ->
log_long_message_queue_event(pid, type)
end)
{:noreply, state}
end
defp log_long_message_queue_event(pid, type) do
with {:message_queue_len, queue_len} <- Process.info(pid, :message_queue_len) do
:telemetry.execute(@vm_monitor_long_message_queue, %{length: queue_len}, %{
process_type: type
})
end
end
defp maybe_start_long_message_queue_timer(%{long_message_queue_timer: nil} = state) do
# A process whose message queue length exceeds the threshold is likely to be spiraling out of
# control. Therefore we need recheck it quite often to capture the dynamics.
#
# Though there's still no guarantee that the VM will not run out of memory before it
# reaches the next metric export tick.
{:ok, timer} = :timer.send_interval(200, :recheck_message_queues)
%{state | long_message_queue_timer: timer}
end
defp maybe_start_long_message_queue_timer(state), do: state
end