X-Git-Url: https://git.xandkar.net/?a=blobdiff_plain;f=src%2Fbeam_stats_consumer_statsd.erl;h=112bb91ec7004282abb125372807cae99c8d3d02;hb=a8d431d1d7c542f66aeb1c272a5a2edf3cd9130d;hp=09a8e957e9ec04f30fc3d6831b6c3925318a3f7b;hpb=b0ab6ee2c0bea2f50a60d2260fc2e71dfb01320b;p=beam_stats.git diff --git a/src/beam_stats_consumer_statsd.erl b/src/beam_stats_consumer_statsd.erl index 09a8e95..112bb91 100644 --- a/src/beam_stats_consumer_statsd.erl +++ b/src/beam_stats_consumer_statsd.erl @@ -1,6 +1,8 @@ -module(beam_stats_consumer_statsd). -include("include/beam_stats.hrl"). +-include("include/beam_stats_ets_table.hrl"). +-include("beam_stats_logging.hrl"). -behaviour(beam_stats_consumer). @@ -15,17 +17,8 @@ , terminate/1 ]). -%% Tests (to be run by CT): --export( - [ ct_test__beam_stats_to_bins/1 - , ct_test__memory_component_to_statsd_msg/1 - , ct_test__statsd_msg_add_name_prefix/1 - , ct_test__statsd_msg_to_bin/1 - , ct_test__node_id_to_bin/1 - ]). - -type option() :: - {consumption_interval , erlang:time()} + {consumption_interval , non_neg_integer()} | {dst_host , inet:ip_address() | inet:hostname()} | {dst_port , inet:port_number()} | {src_port , inet:port_number()} @@ -65,7 +58,7 @@ %% ============================================================================ -spec init([option()]) -> - {erlang:time(), state()}. + {non_neg_integer(), state()}. init(Options) -> ConsumptionInterval = hope_kv_list:get(Options, consumption_interval, 60000), DstHost = hope_kv_list:get(Options, dst_host, ?DEFAULT_DST_HOST), @@ -99,7 +92,7 @@ terminate(#state{sock=SockOpt}) -> -spec try_to_send(state(), binary()) -> state(). try_to_send(#state{sock=none}=State, _) -> - io:format("error: socket closed~n"), + ?log_error("Sending failed. No socket in state."), % TODO: Maybe schedule retry? State; try_to_send( @@ -114,7 +107,10 @@ try_to_send( of ok -> State ; {error, _}=Error -> - io:format("error: gen_udp:send/4 failed: ~p~n", [Error]), + ?log_error( + "gen_udp:send(~p, ~p, ~p, ~p) -> ~p", + [Sock, DstHost, DstPort, Error] + ), % TODO: Do something with unsent messages? ok = gen_udp:close(Sock), State#state{sock=none} @@ -129,7 +125,7 @@ try_to_connect_if_no_socket(#state{sock=none, src_port=SrcPort}=State) -> of {ok, Sock} -> State#state{sock = {some, Sock}} ; {error, _}=Error -> - io:format("error: gen_udp:open/1 failed: ~p~n", [Error]), + ?log_error("gen_udp:open(~p) -> ~p", [SrcPort, Error]), State#state{sock = none} end. @@ -147,13 +143,104 @@ beam_stats_queue_to_binary(Q) -> beam_stats_to_bins(#beam_stats { node_id = NodeID , memory = Memory + , io_bytes_in = IOBytesIn + , io_bytes_out = IOBytesOut + , context_switches = ContextSwitches + , reductions = Reductions + , run_queue = RunQueue + , ets = ETS } ) -> NodeIDBin = node_id_to_bin(NodeID), - Msgs1 = memory_to_msgs(Memory), + Msgs1 = + [ io_bytes_in_to_msg(IOBytesIn) + , io_bytes_out_to_msg(IOBytesOut) + , context_switches_to_msg(ContextSwitches) + , reductions_to_msg(Reductions) + , run_queue_to_msg(RunQueue) + | memory_to_msgs(Memory) + ] + ++ ets_to_msgs(ETS), Msgs2 = [statsd_msg_add_name_prefix(M, NodeIDBin) || M <- Msgs1], [statsd_msg_to_bin(M) || M <- Msgs2]. +-spec run_queue_to_msg(non_neg_integer()) -> + statsd_msg(). +run_queue_to_msg(RunQueue) -> + #statsd_msg + { name = <<"run_queue">> + , value = RunQueue + , type = gauge + }. + +-spec reductions_to_msg(non_neg_integer()) -> + statsd_msg(). +reductions_to_msg(Reductions) -> + #statsd_msg + { name = <<"reductions">> + , value = Reductions + , type = gauge + }. + +-spec context_switches_to_msg(non_neg_integer()) -> + statsd_msg(). +context_switches_to_msg(ContextSwitches) -> + #statsd_msg + { name = <<"context_switches">> + , value = ContextSwitches + , type = gauge + }. + +-spec io_bytes_in_to_msg(non_neg_integer()) -> + statsd_msg(). +io_bytes_in_to_msg(IOBytesIn) -> + #statsd_msg + { name = <<"io.bytes_in">> + , value = IOBytesIn + , type = gauge + }. + +-spec io_bytes_out_to_msg(non_neg_integer()) -> + statsd_msg(). +io_bytes_out_to_msg(IOBytesOut) -> + #statsd_msg + { name = <<"io.bytes_out">> + , value = IOBytesOut + , type = gauge + }. + +-spec ets_to_msgs(beam_stats_ets:t()) -> + [statsd_msg()]. +ets_to_msgs(PerTableStats) -> + NestedMsgs = lists:map(fun ets_table_to_msgs/1, PerTableStats), + lists:append(NestedMsgs). + +-spec ets_table_to_msgs(beam_stats_ets_table:t()) -> + [statsd_msg()]. +ets_table_to_msgs(#beam_stats_ets_table + { id = ID + , name = Name + , size = Size + , memory = Memory + } +) -> + IDBin = beam_stats_ets_table:id_to_bin(ID), + NameBin = atom_to_binary(Name, latin1), + NameAndID = <>, + SizeMsg = + #statsd_msg + { name = <<"ets_table.size.", NameAndID/binary>> + , value = Size + , type = gauge + }, + MemoryMsg = + #statsd_msg + { name = <<"ets_table.memory.", NameAndID/binary>> + , value = Memory + , type = gauge + }, + [SizeMsg, MemoryMsg]. + -spec memory_to_msgs([{atom(), non_neg_integer()}]) -> [statsd_msg()]. memory_to_msgs(Memory) -> @@ -162,8 +249,9 @@ memory_to_msgs(Memory) -> -spec memory_component_to_statsd_msg({atom(), non_neg_integer()}) -> statsd_msg(). memory_component_to_statsd_msg({MemType, MemSize}) when MemSize >= 0 -> + MemTypeBin = atom_to_binary(MemType, latin1), #statsd_msg - { name = atom_to_binary(MemType, latin1) + { name = <<"memory.", MemTypeBin/binary>> , value = MemSize , type = gauge }. @@ -204,28 +292,3 @@ metric_type_to_bin(gauge) -> node_id_to_bin(NodeID) -> NodeIDBin = atom_to_binary(NodeID, utf8), re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]). - -%% ============================================================================ -%% Tests -%% ============================================================================ - -ct_test__beam_stats_to_bins(_Cfg) -> - BEAMStats = #beam_stats{node_id = 'node@host.local', memory = [{foo,1}]}, - [<>] = - beam_stats_to_bins(BEAMStats). - -ct_test__memory_component_to_statsd_msg(_Cfg) -> - #statsd_msg{name = <<"foo">>, value = 1, type = gauge} = - memory_component_to_statsd_msg({foo, 1}). - -ct_test__statsd_msg_add_name_prefix(_Cfg) -> - Msg1 = #statsd_msg{name = <<"foo">>, value = 1, type = gauge}, - Msg2 = statsd_msg_add_name_prefix(Msg1, <<"bar">>), - <> = Msg2#statsd_msg.name. - -ct_test__statsd_msg_to_bin(_Cfg) -> - Msg = #statsd_msg{name = <<"foo">>, value = 1, type = gauge}, - <<"foo:1|g\n">> = statsd_msg_to_bin(Msg). - -ct_test__node_id_to_bin(_Cfg) -> - <<"foo_bar_local">> = node_id_to_bin('foo@bar.local').