, reductions :: non_neg_integer()
, run_queue :: non_neg_integer()
, ets :: beam_stats_ets:t()
+ , processes :: beam_stats_processes:t()
%, statistics :: [{atom() , term()}]
%, system :: [{atom() , term()}]
- %, process :: [{atom() , term()}]
%, port :: [{atom() , term()}]
%, dets :: [{atom() , term()}]
}).
--- /dev/null
+-record(beam_stats_process,
+ { pid :: pid()
+ , registered_name :: hope_option:t(atom())
+ , ancestry :: beam_stats_process:ancestry()
+ , status :: beam_stats_process:status()
+ , memory :: non_neg_integer()
+ , total_heap_size :: non_neg_integer()
+ , stack_size :: non_neg_integer()
+ , message_queue_len :: non_neg_integer()
+ }).
--- /dev/null
+-record(beam_stats_process_ancestry,
+ { raw_initial_call :: mfa()
+ , otp_initial_call :: hope_option:t(mfa())
+ , otp_ancestors :: hope_option:t([atom() | pid()])
+ }).
--- /dev/null
+-record(beam_stats_processes,
+ { individual_stats :: [beam_stats_process:t()]
+ , count_all :: non_neg_integer()
+ , count_exiting :: non_neg_integer()
+ , count_garbage_collecting :: non_neg_integer()
+ , count_registered :: non_neg_integer()
+ , count_runnable :: non_neg_integer()
+ , count_running :: non_neg_integer()
+ , count_suspended :: non_neg_integer()
+ , count_waiting :: non_neg_integer()
+ }).
{application, beam_stats,
[
{description, "Periodic VM stats production and consumption."},
- {vsn, "0.6.0"},
+ {vsn, "0.7.0"},
{registered, []},
{applications,
[ kernel
-include("include/beam_stats.hrl").
-include("include/beam_stats_ets_table.hrl").
+-include("include/beam_stats_process.hrl").
+-include("include/beam_stats_process_ancestry.hrl").
+-include("include/beam_stats_processes.hrl").
-include("beam_stats_logging.hrl").
-behaviour(beam_stats_consumer).
, reductions = Reductions
, run_queue = RunQueue
, ets = ETS
+ , processes = Processes
}
) ->
NodeIDBin = node_id_to_bin(NodeID),
, run_queue_to_msg(RunQueue)
| memory_to_msgs(Memory)
]
- ++ ets_to_msgs(ETS),
+ ++ ets_to_msgs(ETS)
+ ++ procs_to_msgs(Processes),
Msgs2 = [statsd_msg_add_name_prefix(M, NodeIDBin) || M <- Msgs1],
[statsd_msg_to_bin(M) || M <- Msgs2].
, type = gauge
}.
+-spec procs_to_msgs(beam_stats_processes:t()) ->
+ [statsd_msg()].
+procs_to_msgs(
+ #beam_stats_processes
+ { individual_stats = Procs
+ , count_all = CountAll
+ , count_exiting = CountExiting
+ , count_garbage_collecting = CountGarbageCollecting
+ , count_registered = CountRegistered
+ , count_runnable = CountRunnable
+ , count_running = CountRunning
+ , count_suspended = CountSuspended
+ , count_waiting = CountWaiting
+ }
+) ->
+ [ gauge(<<"processes_count_all">> , CountAll)
+ , gauge(<<"processes_count_exiting">> , CountExiting)
+ , gauge(<<"processes_count_garbage_collecting">>, CountGarbageCollecting)
+ , gauge(<<"processes_count_registered">> , CountRegistered)
+ , gauge(<<"processes_count_runnable">> , CountRunnable)
+ , gauge(<<"processes_count_running">> , CountRunning)
+ , gauge(<<"processes_count_suspended">> , CountSuspended)
+ , gauge(<<"processes_count_waiting">> , CountWaiting)
+ | lists:append([proc_to_msgs(P) || P <- Procs])
+ ].
+
+-spec proc_to_msgs(beam_stats_process:t()) ->
+ [statsd_msg()].
+proc_to_msgs(
+ #beam_stats_process
+ { pid = Pid
+ , memory = Memory
+ , total_heap_size = TotalHeapSize
+ , stack_size = StackSize
+ , message_queue_len = MsgQueueLen
+ }=Process
+) ->
+ Origin = beam_stats_process:get_best_known_origin(Process),
+ OriginBin = proc_origin_to_bin(Origin),
+ PidBin = pid_to_bin(Pid),
+ OriginDotPid = <<OriginBin/binary, ".", PidBin/binary>>,
+ [ gauge(<<"process_memory." , OriginDotPid/binary>>, Memory)
+ , gauge(<<"process_total_heap_size." , OriginDotPid/binary>>, TotalHeapSize)
+ , gauge(<<"process_stack_size." , OriginDotPid/binary>>, StackSize)
+ , gauge(<<"process_message_queue_len." , OriginDotPid/binary>>, MsgQueueLen)
+ ].
+
+-spec proc_origin_to_bin(beam_stats_process:best_known_origin()) ->
+ binary().
+proc_origin_to_bin({registered_name, Name}) ->
+ atom_to_binary(Name, utf8);
+proc_origin_to_bin({ancestry, Ancestry}) ->
+ #beam_stats_process_ancestry
+ { raw_initial_call = InitCallRaw
+ , otp_initial_call = InitCallOTPOpt
+ , otp_ancestors = AncestorsOpt
+ } = Ancestry,
+ Blank = <<"NONE">>,
+ InitCallOTPBinOpt = hope_option:map(InitCallOTPOpt , fun mfa_to_bin/1),
+ InitCallOTPBin = hope_option:get(InitCallOTPBinOpt, Blank),
+ AncestorsBinOpt = hope_option:map(AncestorsOpt , fun ancestors_to_bin/1),
+ AncestorsBin = hope_option:get(AncestorsBinOpt , Blank),
+ InitCallRawBin = mfa_to_bin(InitCallRaw),
+ << InitCallRawBin/binary
+ , "--"
+ , InitCallOTPBin/binary
+ , "--"
+ , AncestorsBin/binary
+ >>.
+
+ancestors_to_bin([]) ->
+ <<>>;
+ancestors_to_bin([A | Ancestors]) ->
+ ABin = ancestor_to_bin(A),
+ case ancestors_to_bin(Ancestors)
+ of <<>> ->
+ ABin
+ ; <<AncestorsBin/binary>> ->
+ <<ABin/binary, "-", AncestorsBin/binary>>
+ end.
+
+ancestor_to_bin(A) when is_atom(A) ->
+ atom_to_binary(A, utf8);
+ancestor_to_bin(A) when is_pid(A) ->
+ pid_to_bin(A).
+
+pid_to_bin(Pid) ->
+ PidList = erlang:pid_to_list(Pid),
+ PidBin = re:replace(PidList, "[\.]", "_", [global, {return, binary}]),
+ re:replace(PidBin , "[><]", "" , [global, {return, binary}]).
+
+-spec mfa_to_bin(mfa()) ->
+ binary().
+mfa_to_bin({Module, Function, Arity}) ->
+ ModuleBin = atom_to_binary(Module , utf8),
+ FunctionBin = atom_to_binary(Function, utf8),
+ ArityBin = erlang:integer_to_binary(Arity),
+ <<ModuleBin/binary, "-", FunctionBin/binary, "-", ArityBin/binary>>.
+
+
+-spec gauge(binary(), integer()) ->
+ statsd_msg().
+gauge(<<Name/binary>>, Value) when is_integer(Value) ->
+ #statsd_msg
+ { name = Name
+ , value = Value
+ , type = gauge
+ }.
+
-spec ets_to_msgs(beam_stats_ets:t()) ->
[statsd_msg()].
ets_to_msgs(PerTableStats) ->
--- /dev/null
+-module(beam_stats_process).
+
+-include("include/beam_stats_process_ancestry.hrl").
+-include("include/beam_stats_process.hrl").
+
+-export_type(
+ [ t/0
+ , status/0
+ , ancestry/0
+ , best_known_origin/0
+ ]).
+
+-export(
+ [ of_pid/1
+ , get_best_known_origin/1
+ , print/1
+ ]).
+
+-type status() ::
+ exiting
+ | garbage_collecting
+ | runnable
+ | running
+ | suspended
+ | waiting
+ .
+
+-type ancestry() ::
+ #beam_stats_process_ancestry{}.
+
+-type best_known_origin() ::
+ {registered_name , atom()}
+ | {ancestry , ancestry()}
+ .
+
+-define(T, #?MODULE).
+
+-type t() ::
+ ?T{}.
+
+%% ============================================================================
+%% Public API
+%% ============================================================================
+
+-spec of_pid(pid()) ->
+ t().
+of_pid(Pid) ->
+ Dict = pid_info_exn(Pid, dictionary),
+ Ancestry =
+ #beam_stats_process_ancestry
+ { raw_initial_call = pid_info_exn(Pid, initial_call)
+ , otp_initial_call = hope_kv_list:get(Dict, '$initial_call')
+ , otp_ancestors = hope_kv_list:get(Dict, '$ancestors')
+ },
+ ?T
+ { pid = Pid
+ , registered_name = pid_info_opt(Pid, registered_name)
+ , ancestry = Ancestry
+ , status = pid_info_exn(Pid, status)
+ , memory = pid_info_exn(Pid, memory)
+ , total_heap_size = pid_info_exn(Pid, total_heap_size)
+ , stack_size = pid_info_exn(Pid, stack_size)
+ , message_queue_len = pid_info_exn(Pid, message_queue_len)
+ }.
+
+-spec print(t()) ->
+ ok.
+print(
+ ?T
+ { pid = Pid
+ , registered_name = RegisteredNameOpt
+ , ancestry = #beam_stats_process_ancestry
+ { raw_initial_call = InitialCallRaw
+ , otp_initial_call = InitialCallOTPOpt
+ , otp_ancestors = AncestorsOpt
+ }
+ , status = Status
+ , memory = Memory
+ , total_heap_size = TotalHeapSize
+ , stack_size = StackSize
+ , message_queue_len = MsgQueueLen
+ }=T
+) ->
+ BestKnownOrigin = get_best_known_origin(T),
+ io:format("--------------------------------------------------~n"),
+ io:format(
+ "Pid : ~p~n"
+ "BestKnownOrigin : ~p~n"
+ "RegisteredNameOpt : ~p~n"
+ "InitialCallRaw : ~p~n"
+ "InitialCallOTPOpt : ~p~n"
+ "AncestorsOpt : ~p~n"
+ "Status : ~p~n"
+ "Memory : ~p~n"
+ "TotalHeapSize : ~p~n"
+ "StackSize : ~p~n"
+ "MsgQueueLen : ~p~n"
+ "~n",
+ [ Pid
+ , BestKnownOrigin
+ , RegisteredNameOpt
+ , InitialCallRaw
+ , InitialCallOTPOpt
+ , AncestorsOpt
+ , Status
+ , Memory
+ , TotalHeapSize
+ , StackSize
+ , MsgQueueLen
+ ]
+ ).
+
+%% ============================================================================
+%% Private helpers
+%% ============================================================================
+
+pid_info_exn(Pid, Key) ->
+ {some, Value} = pid_info_opt(Pid, Key),
+ Value.
+
+pid_info_opt(Pid, Key) ->
+ case {Key, erlang:process_info(Pid, Key)}
+ of {registered_name, []} -> none
+ ; {_ , {Key, Value}} -> {some, Value}
+ end.
+
+%% ============================================================================
+%% Process code origin approximation or naming the anonymous processes.
+%%
+%% At runtime, given a PID, how precicely can we identify the origin of the
+%% code it is running?
+%%
+%% We have these data points:
+%%
+%% - Sometimes | registered name (if so, we're done)
+%% - Sometimes | ancestor PIDs or registered names
+%% - Always | initial_call (can be too generic, such as erlang:apply)
+%% - Always | current_function (can be too far down the stack)
+%% - Always | current_location (can be too far down the stack)
+%% - Potentially | application tree, but maybe expensive to compute, need to check
+%% ============================================================================
+
+-define(TAG(Tag), fun (X) -> {Tag, X} end).
+
+-spec get_best_known_origin(t()) ->
+ best_known_origin().
+get_best_known_origin(?T{registered_name={some, RegisteredName}}) ->
+ {registered_name, RegisteredName};
+get_best_known_origin(?T{registered_name=none, ancestry=Ancestry}) ->
+ {ancestry, Ancestry}.
--- /dev/null
+-module(beam_stats_processes).
+
+-include("include/beam_stats_process.hrl").
+-include("include/beam_stats_processes.hrl").
+
+-export_type(
+ [ t/0
+ ]).
+
+-export(
+ [ collect/0
+ , collect_and_print/0
+ , print/1
+ ]).
+
+-define(T, #?MODULE).
+
+-type t() ::
+ ?T{}.
+
+-spec collect() ->
+ t().
+collect() ->
+ Ps = [beam_stats_process:of_pid(P) || P <- erlang:processes()],
+ ?T
+ { individual_stats
+ = Ps
+ , count_all
+ = length(Ps)
+ , count_exiting
+ = length([P || P <- Ps, P#beam_stats_process.status =:= exiting])
+ , count_garbage_collecting
+ = length([P || P <- Ps, P#beam_stats_process.status =:= garbage_collecting])
+ , count_registered
+ = length(registered())
+ , count_runnable
+ = length([P || P <- Ps, P#beam_stats_process.status =:= runnable])
+ , count_running
+ = length([P || P <- Ps, P#beam_stats_process.status =:= running])
+ , count_suspended
+ = length([P || P <- Ps, P#beam_stats_process.status =:= suspended])
+ , count_waiting
+ = length([P || P <- Ps, P#beam_stats_process.status =:= waiting])
+ }.
+
+collect_and_print() ->
+ print(collect()).
+
+-spec print(t()) ->
+ ok.
+print(
+ ?T
+ { individual_stats = PerProcessStats
+ , count_all = CountAll
+ , count_exiting = CountExiting
+ , count_garbage_collecting = CountGarbageCollecting
+ , count_registered = CountRegistered
+ , count_runnable = CountRunnable
+ , count_running = CountRunning
+ , count_suspended = CountSuspended
+ , count_waiting = CountWaiting
+ }
+) ->
+ PerProcessStatsSorted = lists:sort(
+ fun (#beam_stats_process{memory=A}, #beam_stats_process{memory=B}) ->
+ % From lowest to highest:
+ A < B
+ end,
+ PerProcessStats
+ ),
+ lists:foreach(fun beam_stats_process:print/1, PerProcessStatsSorted),
+ io:format("==================================================~n"),
+ io:format(
+ "CountAll : ~b~n"
+ "CountExiting : ~b~n"
+ "CountGarbageCollecting : ~b~n"
+ "CountRegistered : ~b~n"
+ "CountRunnable : ~b~n"
+ "CountRunning : ~b~n"
+ "CountSuspended : ~b~n"
+ "CountWaiting : ~b~n"
+ "~n",
+ [ CountAll
+ , CountExiting
+ , CountGarbageCollecting
+ , CountRegistered
+ , CountRunnable
+ , CountRunning
+ , CountSuspended
+ , CountWaiting
+ ]
+ ).
-module(beam_stats_state).
-include("include/beam_stats.hrl").
+-include("include/beam_stats_process.hrl").
+-include("include/beam_stats_processes.hrl").
-export_type(
[ t/0
-record(snapshots,
{ memory :: [{atom(), non_neg_integer()}]
+ , processes :: beam_stats_processes:t()
, run_queue :: non_neg_integer()
, ets :: beam_stats_ets:t()
}).
, snapshots =
#snapshots
{ memory = Memory
+ , processes = Processes
, run_queue = RunQueue
, ets = ETS
}
, reductions = Reductions
, run_queue = RunQueue
, ets = ETS
+ , processes = Processes
}.
snapshots_new() ->
#snapshots
{ memory = erlang:memory()
+ , processes = beam_stats_processes:collect()
, run_queue = erlang:statistics(run_queue)
, ets = beam_stats_ets:collect()
}.
-include_lib("beam_stats/include/beam_stats.hrl").
-include_lib("beam_stats/include/beam_stats_ets_table.hrl").
+-include_lib("beam_stats/include/beam_stats_process.hrl").
+-include_lib("beam_stats/include/beam_stats_process_ancestry.hrl").
+-include_lib("beam_stats/include/beam_stats_processes.hrl").
-export(
[ all/0
Properties = [],
[{?GROUP, Properties, Tests}].
-%% =============================================================================
+%% ============================================================================
%% Test cases
-%% =============================================================================
+%% ============================================================================
t_send(_Cfg) ->
+ Pid0 = list_to_pid("<0.0.0>"),
+ Pid1 = list_to_pid("<0.1.0>"),
+ Pid2 = list_to_pid("<0.2.0>"),
+ Pid3 = list_to_pid("<0.3.0>"),
+ Process1 =
+ #beam_stats_process
+ { pid = Pid1
+ , registered_name = {some, reg_name_foo}
+ , ancestry =
+ #beam_stats_process_ancestry
+ { raw_initial_call = {foo_mod, foo_fun, 2}
+ , otp_initial_call = none
+ , otp_ancestors = none
+ }
+ , status = running
+ , memory = 15
+ , total_heap_size = 25
+ , stack_size = 10
+ , message_queue_len = 0
+ },
+ Process2 =
+ #beam_stats_process
+ { pid = Pid2
+ , registered_name = none
+ , ancestry =
+ #beam_stats_process_ancestry
+ { raw_initial_call = {bar_mod, bar_fun, 1}
+ , otp_initial_call = none
+ , otp_ancestors = none
+ }
+ , status = running
+ , memory = 25
+ , total_heap_size = 35
+ , stack_size = 40
+ , message_queue_len = 5
+ },
+ Process3 =
+ #beam_stats_process
+ { pid = Pid3
+ , registered_name = none
+ , ancestry =
+ #beam_stats_process_ancestry
+ { raw_initial_call = {baz_mod, baz_fun, 3}
+ , otp_initial_call = {some, {baz_otp_mod, baz_otp_fun, 2}}
+ , otp_ancestors = {some, [Pid0, Pid1]}
+ }
+ , status = running
+ , memory = 25
+ , total_heap_size = 35
+ , stack_size = 40
+ , message_queue_len = 1
+ },
+ Processes =
+ #beam_stats_processes
+ { individual_stats =
+ [ Process1
+ , Process2
+ , Process3
+ ]
+ , count_all = 3
+ , count_exiting = 0
+ , count_garbage_collecting = 0
+ , count_registered = 1
+ , count_runnable = 0
+ , count_running = 3
+ , count_suspended = 0
+ , count_waiting = 0
+ },
ETSTableStatsFoo =
#beam_stats_ets_table
{ id = foo
, reductions = 9
, run_queue = 17
, ets = [ETSTableStatsFoo, ETSTableStatsBar]
+ , processes = Processes
},
ServerPort = 8125,
{ok, ServerSocket} = gen_udp:open(ServerPort, [binary, {active, false}]),
State2 = beam_stats_consumer_statsd:consume(BEAMStatsQ, State1),
{} = beam_stats_consumer_statsd:terminate(State2),
ResultOfReceive1 = gen_udp:recv(ServerSocket, 0),
- {ok, {_, _, PacketReceived1}} = ResultOfReceive1,
ResultOfReceive2 = gen_udp:recv(ServerSocket, 0),
- {ok, {_, _, PacketReceived2}} = ResultOfReceive2,
+ ResultOfReceive3 = gen_udp:recv(ServerSocket, 0),
+ ResultOfReceive4 = gen_udp:recv(ServerSocket, 0),
ok = gen_udp:close(ServerSocket),
+ {ok, {_, _, PacketReceived1}} = ResultOfReceive1,
+ {ok, {_, _, PacketReceived2}} = ResultOfReceive2,
+ {ok, {_, _, PacketReceived3}} = ResultOfReceive3,
+ {ok, {_, _, PacketReceived4}} = ResultOfReceive4,
ct:log("PacketReceived1: ~n~s~n", [PacketReceived1]),
ct:log("PacketReceived2: ~n~s~n", [PacketReceived2]),
- PacketsCombined = <<PacketReceived1/binary, PacketReceived2/binary>>,
+ ct:log("PacketReceived3: ~n~s~n", [PacketReceived3]),
+ ct:log("PacketReceived4: ~n~s~n", [PacketReceived4]),
+ PacketsCombined =
+ << PacketReceived1/binary
+ , PacketReceived2/binary
+ , PacketReceived3/binary
+ , PacketReceived4/binary
+ >>,
ct:log("PacketsCombined: ~n~s~n", [PacketsCombined]),
MsgsExpected =
[ <<"beam_stats.node_foo_host_bar.io.bytes_in:3|g">>
, <<"beam_stats.node_foo_host_bar.ets_table.memory.foo.foo:25|g">>
, <<"beam_stats.node_foo_host_bar.ets_table.size.bar.37:8|g">>
, <<"beam_stats.node_foo_host_bar.ets_table.memory.bar.37:38|g">>
+
+ % Processes totals
+ , <<"beam_stats.node_foo_host_bar.processes_count_all:3|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_exiting:0|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_garbage_collecting:0|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_registered:1|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_runnable:0|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_running:3|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_suspended:0|g">>
+ , <<"beam_stats.node_foo_host_bar.processes_count_waiting:0|g">>
+
+ % Process 1
+ , <<"beam_stats.node_foo_host_bar.process_memory.reg_name_foo.0_1_0:15|g">>
+ , <<"beam_stats.node_foo_host_bar.process_total_heap_size.reg_name_foo.0_1_0:25|g">>
+ , <<"beam_stats.node_foo_host_bar.process_stack_size.reg_name_foo.0_1_0:10|g">>
+ , <<"beam_stats.node_foo_host_bar.process_message_queue_len.reg_name_foo.0_1_0:0|g">>
+
+ % Process 2
+ , <<"beam_stats.node_foo_host_bar.process_memory.bar_mod-bar_fun-1--NONE--NONE.0_2_0:25|g">>
+ , <<"beam_stats.node_foo_host_bar.process_total_heap_size.bar_mod-bar_fun-1--NONE--NONE.0_2_0:35|g">>
+ , <<"beam_stats.node_foo_host_bar.process_stack_size.bar_mod-bar_fun-1--NONE--NONE.0_2_0:40|g">>
+ , <<"beam_stats.node_foo_host_bar.process_message_queue_len.bar_mod-bar_fun-1--NONE--NONE.0_2_0:5|g">>
+
+ % Process 3
+ , <<"beam_stats.node_foo_host_bar.process_memory.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:25|g">>
+ , <<"beam_stats.node_foo_host_bar.process_total_heap_size.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:35|g">>
+ , <<"beam_stats.node_foo_host_bar.process_stack_size.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:40|g">>
+ , <<"beam_stats.node_foo_host_bar.process_message_queue_len.baz_mod-baz_fun-3--baz_otp_mod-baz_otp_fun-2--0_0_0-0_1_0.0_3_0:1|g">>
],
MsgsReceived = binary:split(PacketsCombined, <<"\n">>, [global, trim]),
RemoveExpectedFromReceived =
home / code / beam_stats.git / commitdiff
