1 -module(beam_stats_consumer_statsd).
3 -include("include/beam_stats.hrl").
4 -include("include/beam_stats_ets_table.hrl").
5 -include("beam_stats_logging.hrl").
7 -behaviour(beam_stats_consumer).
21 {consumption_interval , non_neg_integer()}
22 | {dst_host , inet:ip_address() | inet:hostname()}
23 | {dst_port , inet:port_number()}
24 | {src_port , inet:port_number()}
27 -define(DEFAULT_DST_HOST, "localhost").
28 -define(DEFAULT_DST_PORT, 8125).
29 -define(DEFAULT_SRC_PORT, 8124).
31 -type metric_type() ::
32 % TODO: Add other metric types
37 , value :: non_neg_integer()
38 , type :: metric_type()
45 { sock :: hope_option:t(gen_udp:socket())
46 , dst_host :: inet:ip_address() | inet:hostname()
47 , dst_port :: inet:port_number()
48 , src_port :: inet:port_number()
54 -define(PATH_PREFIX, "beam_stats").
56 %% ============================================================================
57 %% Consumer implementation
58 %% ============================================================================
60 -spec init([option()]) ->
61 {non_neg_integer(), state()}.
63 ConsumptionInterval = hope_kv_list:get(Options, consumption_interval, 60000),
64 DstHost = hope_kv_list:get(Options, dst_host, ?DEFAULT_DST_HOST),
65 DstPort = hope_kv_list:get(Options, dst_port, ?DEFAULT_DST_PORT),
66 SrcPort = hope_kv_list:get(Options, src_port, ?DEFAULT_SRC_PORT),
73 {ConsumptionInterval, State}.
75 -spec consume(beam_stats_consumer:queue(), state()) ->
77 consume(Q, #state{}=State1) ->
78 Payload = beam_stats_queue_to_binary(Q),
79 State2 = try_to_connect_if_no_socket(State1),
80 try_to_send(State2, Payload).
82 -spec terminate(state()) ->
84 terminate(#state{sock=SockOpt}) ->
85 ok = hope_option:iter(SockOpt, fun gen_udp:close/1),
88 %% ============================================================================
90 %% ============================================================================
92 -spec try_to_send(state(), binary()) ->
94 try_to_send(#state{sock=none}=State, _) ->
95 ?log_error("Sending failed. No socket in state."),
96 % TODO: Maybe schedule retry?
100 { sock = {some, Sock}
106 case gen_udp:send(Sock, DstHost, DstPort, Payload)
109 ; {error, _}=Error ->
111 "gen_udp:send(~p, ~p, ~p, ~p) -> ~p",
112 [Sock, DstHost, DstPort, Error]
114 % TODO: Do something with unsent messages?
115 ok = gen_udp:close(Sock),
116 State#state{sock=none}
119 -spec try_to_connect_if_no_socket(state()) ->
121 try_to_connect_if_no_socket(#state{sock={some, _}}=State) ->
123 try_to_connect_if_no_socket(#state{sock=none, src_port=SrcPort}=State) ->
124 case gen_udp:open(SrcPort)
126 State#state{sock = {some, Sock}}
127 ; {error, _}=Error ->
128 ?log_error("gen_udp:open(~p) -> ~p", [SrcPort, Error]),
129 State#state{sock = none}
132 %% ============================================================================
134 %% ============================================================================
136 -spec beam_stats_queue_to_binary(beam_stats_consumer:queue()) ->
138 beam_stats_queue_to_binary(Q) ->
139 iolist_to_binary([beam_stats_to_bins(B) || B <- queue:to_list(Q)]).
141 -spec beam_stats_to_bins(beam_stats:t()) ->
143 beam_stats_to_bins(#beam_stats
146 , io_bytes_in = IOBytesIn
147 , io_bytes_out = IOBytesOut
148 , context_switches = ContextSwitches
149 , reductions = Reductions
150 , run_queue = RunQueue
154 NodeIDBin = node_id_to_bin(NodeID),
156 [ io_bytes_in_to_msg(IOBytesIn)
157 , io_bytes_out_to_msg(IOBytesOut)
158 , context_switches_to_msg(ContextSwitches)
159 , reductions_to_msg(Reductions)
160 , run_queue_to_msg(RunQueue)
161 | memory_to_msgs(Memory)
164 Msgs2 = [statsd_msg_add_name_prefix(M, NodeIDBin) || M <- Msgs1],
165 [statsd_msg_to_bin(M) || M <- Msgs2].
167 -spec run_queue_to_msg(non_neg_integer()) ->
169 run_queue_to_msg(RunQueue) ->
171 { name = <<"run_queue">>
176 -spec reductions_to_msg(non_neg_integer()) ->
178 reductions_to_msg(Reductions) ->
180 { name = <<"reductions">>
185 -spec context_switches_to_msg(non_neg_integer()) ->
187 context_switches_to_msg(ContextSwitches) ->
189 { name = <<"context_switches">>
190 , value = ContextSwitches
194 -spec io_bytes_in_to_msg(non_neg_integer()) ->
196 io_bytes_in_to_msg(IOBytesIn) ->
198 { name = <<"io.bytes_in">>
203 -spec io_bytes_out_to_msg(non_neg_integer()) ->
205 io_bytes_out_to_msg(IOBytesOut) ->
207 { name = <<"io.bytes_out">>
212 -spec ets_to_msgs(beam_stats_ets:t()) ->
214 ets_to_msgs(PerTableStats) ->
215 NestedMsgs = lists:map(fun ets_table_to_msgs/1, PerTableStats),
216 lists:append(NestedMsgs).
218 -spec ets_table_to_msgs(beam_stats_ets_table:t()) ->
220 ets_table_to_msgs(#beam_stats_ets_table
227 IDBin = beam_stats_ets_table:id_to_bin(ID),
228 NameBin = atom_to_binary(Name, latin1),
229 NameAndID = <<NameBin/binary, ".", IDBin/binary>>,
232 { name = <<"ets_table.size.", NameAndID/binary>>
238 { name = <<"ets_table.memory.", NameAndID/binary>>
242 [SizeMsg, MemoryMsg].
244 -spec memory_to_msgs([{atom(), non_neg_integer()}]) ->
246 memory_to_msgs(Memory) ->
247 [memory_component_to_statsd_msg(MC) || MC <- Memory].
249 -spec memory_component_to_statsd_msg({atom(), non_neg_integer()}) ->
251 memory_component_to_statsd_msg({MemType, MemSize}) when MemSize >= 0 ->
252 MemTypeBin = atom_to_binary(MemType, latin1),
254 { name = <<"memory.", MemTypeBin/binary>>
259 -spec statsd_msg_add_name_prefix(statsd_msg(), binary()) ->
261 statsd_msg_add_name_prefix(#statsd_msg{name=Name1}=Msg, <<NodeID/binary>>) ->
262 Prefix = <<?PATH_PREFIX, ".", NodeID/binary, ".">>,
263 Name2 = <<Prefix/binary, Name1/binary>>,
264 Msg#statsd_msg{name=Name2}.
266 -spec statsd_msg_to_bin(statsd_msg()) ->
270 { name = <<Name/binary>>
272 , type = Type = gauge
275 TypeBin = metric_type_to_bin(Type),
276 ValueBin = integer_to_binary(Value),
285 -spec metric_type_to_bin(metric_type()) ->
287 metric_type_to_bin(gauge) ->
290 -spec node_id_to_bin(node()) ->
292 node_id_to_bin(NodeID) ->
293 NodeIDBin = atom_to_binary(NodeID, utf8),
294 re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]).