Send per-process metrics to StatsD

diff --git a/include/beam_stats_process_ancestry.hrl b/include/beam_stats_process_ancestry.hrl
index 45ae9e4..9315f65 100644 (file)
--- a/include/beam_stats_process_ancestry.hrl
+++ b/include/beam_stats_process_ancestry.hrl
@@ -1,5 +1,5 @@
 -record(beam_stats_process_ancestry,
     { raw_initial_call  ::               mfa()
     , otp_initial_call  :: hope_option:t(mfa())
-    , otp_ancestors     ::               [{name, atom()} | {call, mfa()}]
+    , otp_ancestors     :: hope_option:t([atom() | pid()])
     }).

diff --git a/src/beam_stats_consumer_statsd.erl b/src/beam_stats_consumer_statsd.erl
index 1ce0afb..8b0b437 100644 (file)
--- a/src/beam_stats_consumer_statsd.erl
+++ b/src/beam_stats_consumer_statsd.erl
@@ -2,6 +2,9 @@
 
 -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).
@@ -160,6 +163,7 @@ beam_stats_to_bins(#beam_stats
     , reductions       = Reductions
     , run_queue        = RunQueue
     , ets              = ETS
+    , processes        = Processes
     }
 ) ->
     NodeIDBin = node_id_to_bin(NodeID),
@@ -171,7 +175,8 @@ beam_stats_to_bins(#beam_stats
         , 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].
 
@@ -220,6 +225,115 @@ io_bytes_out_to_msg(IOBytesOut) ->
     , 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) ->

diff --git a/test/beam_stats_consumer_statsd_SUITE.erl b/test/beam_stats_consumer_statsd_SUITE.erl
index 20a2de7..ffd0c46 100644 (file)
--- a/test/beam_stats_consumer_statsd_SUITE.erl
+++ b/test/beam_stats_consumer_statsd_SUITE.erl
@@ -2,6 +2,9 @@
 
 -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
@@ -29,11 +32,79 @@ groups() ->
     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
@@ -59,6 +130,7 @@ t_send(_Cfg) ->
     , reductions       = 9
     , run_queue        = 17
     , ets              = [ETSTableStatsFoo, ETSTableStatsBar]
+    , processes        = Processes
     },
     ServerPort = 8125,
     {ok, ServerSocket} = gen_udp:open(ServerPort, [binary, {active, false}]),
@@ -68,13 +140,24 @@ t_send(_Cfg) ->
     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">>
@@ -89,6 +172,34 @@ t_send(_Cfg) ->
         , <<"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 =