From: Siraaj Khandkar Date: Fri, 18 Sep 2015 16:04:36 +0000 (-0400) Subject: Send per-process metrics to StatsD X-Git-Tag: 0.7.0^2~1 X-Git-Url: https://git.xandkar.net/?a=commitdiff_plain;h=69ebab9707aaa0fbe576c3f0d7f0b731ee55ce7d;p=beam_stats.git 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 --- 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 --- 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 = <>, + [ 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 + ; <> -> + <> + 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), + <>. + + +-spec gauge(binary(), integer()) -> + statsd_msg(). +gauge(<>, 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 --- 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 = <>, + 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 =