-module(beam_stats_consumer_statsd).
-include("include/beam_stats.hrl").
+-include("beam_stats_logging.hrl").
-behaviour(beam_stats_consumer).
, 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()}
+ | {num_msgs_per_packet , non_neg_integer()}
+ | {static_node_name , binary()}
.
-define(DEFAULT_DST_HOST, "localhost").
-define(DEFAULT_DST_PORT, 8125).
-define(DEFAULT_SRC_PORT, 8124).
--type metric_type() ::
- % TODO: Add other metric types
- gauge.
-
--record(statsd_msg,
- { name :: binary()
- , value :: non_neg_integer()
- , type :: metric_type()
- }).
-
--type statsd_msg() ::
- #statsd_msg{}.
-
-record(state,
{ sock :: hope_option:t(gen_udp:socket())
, dst_host :: inet:ip_address() | inet:hostname()
, dst_port :: inet:port_number()
, src_port :: inet:port_number()
+ , num_msgs_per_packet :: non_neg_integer()
+ , static_node_name :: hope_option:t(binary())
}).
-type state() ::
#state{}.
--define(PATH_PREFIX, "beam_stats").
-
%% ============================================================================
%% Consumer implementation
%% ============================================================================
-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),
DstPort = hope_kv_list:get(Options, dst_port, ?DEFAULT_DST_PORT),
SrcPort = hope_kv_list:get(Options, src_port, ?DEFAULT_SRC_PORT),
+ NumMsgsPerPacket = hope_kv_list:get(Options, num_msgs_per_packet, 10),
+ StaticNodeNameOpt = hope_kv_list:get(Options, static_node_name),
State = #state
{ sock = none
, dst_host = DstHost
, dst_port = DstPort
, src_port = SrcPort
+ , num_msgs_per_packet = NumMsgsPerPacket
+ , static_node_name = StaticNodeNameOpt
},
{ConsumptionInterval, State}.
-spec consume(beam_stats_consumer:queue(), state()) ->
state().
-consume(Q, #state{}=State1) ->
- Payload = beam_stats_queue_to_binary(Q),
- State2 = try_to_connect_if_no_socket(State1),
- try_to_send(State2, Payload).
+consume(
+ Q,
+ #state
+ { num_msgs_per_packet = NumMsgsPerPacket
+ , static_node_name = StaticNodeNameOpt
+ }=State
+) ->
+ Packets = beam_stats_queue_to_packets(Q, NumMsgsPerPacket, StaticNodeNameOpt),
+ lists:foldl(fun try_to_connect_and_send/2, State, Packets).
-spec terminate(state()) ->
{}.
terminate(#state{sock=SockOpt}) ->
- ok = hope_option:iter(SockOpt, fun gen_udp:close/1),
- {}.
+ hope_option:iter(SockOpt, fun gen_udp:close/1).
%% ============================================================================
%% Transport
%% ============================================================================
--spec try_to_send(state(), binary()) ->
+-spec try_to_connect_and_send(iolist(), state()) ->
+ state().
+try_to_connect_and_send(Payload, #state{}=State1) ->
+ State2 = try_to_connect_if_no_socket(State1),
+ try_to_send(State2, Payload).
+
+-spec try_to_send(state(), iolist()) ->
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(
of ok ->
State
; {error, _}=Error ->
- io:format("error: gen_udp:send/4 failed: ~p~n", [Error]),
+ ?log_error(
+ "gen_udp:send(~p, ~p, ~p, Payload) -> ~p",
+ [Sock, DstHost, DstPort, Error]
+ ),
% TODO: Do something with unsent messages?
ok = gen_udp:close(Sock),
State#state{sock=none}
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.
%% Serialization
%% ============================================================================
--spec beam_stats_queue_to_binary(beam_stats_consumer:queue()) ->
- binary().
-beam_stats_queue_to_binary(Q) ->
- iolist_to_binary([beam_stats_to_bins(B) || B <- queue:to_list(Q)]).
-
--spec beam_stats_to_bins(beam_stats:t()) ->
- [binary()].
-beam_stats_to_bins(#beam_stats
- { node_id = NodeID
- , memory = Memory
- }
+-spec beam_stats_queue_to_packets(
+ beam_stats_consumer:queue(),
+ non_neg_integer(),
+ hope_option:t(binary())
) ->
- NodeIDBin = node_id_to_bin(NodeID),
- Msgs1 = memory_to_msgs(Memory),
- Msgs2 = [statsd_msg_add_name_prefix(M, NodeIDBin) || M <- Msgs1],
- [statsd_msg_to_bin(M) || M <- Msgs2].
-
--spec memory_to_msgs([{atom(), non_neg_integer()}]) ->
- [statsd_msg()].
-memory_to_msgs(Memory) ->
- [memory_component_to_statsd_msg(MC) || MC <- Memory].
-
--spec memory_component_to_statsd_msg({atom(), non_neg_integer()}) ->
- statsd_msg().
-memory_component_to_statsd_msg({MemType, MemSize}) when MemSize >= 0 ->
- #statsd_msg
- { name = atom_to_binary(MemType, latin1)
- , value = MemSize
- , type = gauge
- }.
-
--spec statsd_msg_add_name_prefix(statsd_msg(), binary()) ->
- statsd_msg().
-statsd_msg_add_name_prefix(#statsd_msg{name=Name1}=Msg, <<NodeID/binary>>) ->
- Prefix = <<?PATH_PREFIX, ".", NodeID/binary, ".">>,
- Name2 = <<Prefix/binary, Name1/binary>>,
- Msg#statsd_msg{name=Name2}.
-
--spec statsd_msg_to_bin(statsd_msg()) ->
- binary().
-statsd_msg_to_bin(
- #statsd_msg
- { name = <<Name/binary>>
- , value = Value
- , type = Type = gauge
- }
-) when Value >= 0 ->
- TypeBin = metric_type_to_bin(Type),
- ValueBin = integer_to_binary(Value),
- << Name/binary
- , ":"
- , ValueBin/binary
- , "|"
- , TypeBin/binary
- , "\n"
- >>.
-
--spec metric_type_to_bin(metric_type()) ->
- binary().
-metric_type_to_bin(gauge) ->
- <<"g">>.
-
--spec node_id_to_bin(node()) ->
- binary().
-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}]},
- [<<?PATH_PREFIX, ".node_host_local.foo:1|g\n">>] =
- 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">>),
- <<?PATH_PREFIX,".bar.foo">> = 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').
+ [iolist()].
+beam_stats_queue_to_packets(Q, NumMsgsPerPacket, StaticNodeNameOpt) ->
+ MsgIOLists = lists:append([beam_stats_to_iolists(B, StaticNodeNameOpt) || B <- queue:to_list(Q)]),
+ hope_list:divide(MsgIOLists, NumMsgsPerPacket).
+
+-spec beam_stats_to_iolists(beam_stats:t(), hope_option:t(binary())) ->
+ [iolist()].
+beam_stats_to_iolists(#beam_stats{node_id=NodeID}=BeamStats, StaticNodeNameOpt) ->
+ NodeIDBinDefault = beam_stats_msg_graphite:node_id_to_bin(NodeID),
+ NodeIDBin = hope_option:get(StaticNodeNameOpt, NodeIDBinDefault),
+ MsgsGraphite = beam_stats_msg_graphite:of_beam_stats(BeamStats, NodeIDBin),
+ MsgsStatsD =
+ lists:map(fun beam_stats_msg_statsd_gauge:of_msg_graphite/1, MsgsGraphite),
+ lists:map(fun beam_stats_msg_statsd_gauge:to_iolist/1, MsgsStatsD).
home / code / beam_stats.git / blobdiff