1 -module(beam_stats_consumer_graphite).
3 -include("include/beam_stats.hrl").
4 -include("beam_stats_logging.hrl").
6 -behaviour(beam_stats_consumer).
19 {consumption_interval , non_neg_integer()}
20 | {host , inet:ip_address() | inet:hostname()}
21 | {port , inet:port_number()}
22 | {timeout , timeout()}
26 { sock = none :: hope_option:t(Socket :: port())
27 , host :: inet:ip_address() | inet:hostname()
28 , port :: inet:port_number()
29 , timeout :: timeout()
35 -define(PATH_PREFIX , <<"beam_stats">>).
36 -define(DEFAULT_HOST , "localhost").
37 -define(DEFAULT_PORT , 2003).
38 -define(DEFAULT_TIMEOUT , 5000).
40 -spec init([option()]) ->
41 {non_neg_integer(), state()}.
43 Get = fun (Key, Default) -> hope_kv_list:get(Options, Key, Default) end,
44 ConsumptionInterval = Get(consumption_interval, 60000),
47 , host = Get(host , ?DEFAULT_HOST)
48 , port = Get(port , ?DEFAULT_PORT)
49 , timeout = Get(timeout , ?DEFAULT_TIMEOUT)
51 {ConsumptionInterval, State}.
53 -spec consume(beam_stats_consumer:queue(), state()) ->
55 consume(Q, #state{}=State1) ->
56 Payload = beam_stats_queue_to_binary(Q),
57 State2 = try_to_connect_if_no_socket(State1),
58 try_to_send(State2, Payload).
60 -spec terminate(state()) ->
62 terminate(#state{sock=SockOpt}) ->
63 ok = hope_option:iter(SockOpt, fun gen_tcp:close/1),
66 %% ============================================================================
68 -spec try_to_send(state(), binary()) ->
70 try_to_send(#state{sock=none}=State, _) ->
71 ?log_error("Sending failed. No socket in state."),
72 % TODO: Maybe schedule retry?
74 try_to_send(#state{sock={some, Sock}}=State, Payload) ->
75 case gen_tcp:send(Sock, Payload)
79 ?log_error("gen_tcp:send(~p, ~p) -> ~p", [Sock, Payload, Error]),
80 % TODO: Maybe schedule retry?
81 ok = gen_tcp:close(Sock),
82 State#state{sock=none}
85 -spec try_to_connect_if_no_socket(state()) ->
87 try_to_connect_if_no_socket(#state{sock={some, _}}=State) ->
89 try_to_connect_if_no_socket(
97 Options = [binary, {active, false}],
98 case gen_tcp:connect(Host, Port, Options, Timeout)
100 State#state{sock = {some, Sock}}
101 ; {error, _}=Error ->
103 "gen_tcp:connect(~p, ~p, ~p, ~p) -> ~p",
104 [Host, Port, Options, Timeout, Error]
106 State#state{sock = none}
109 -spec beam_stats_queue_to_binary(beam_stats_consumer:queue()) ->
111 beam_stats_queue_to_binary(Q) ->
112 Bins = [beam_stats_to_bins(B) || B <- queue:to_list(Q)],
113 iolist_to_binary(Bins).
115 -spec beam_stats_to_bins(beam_stats:t()) ->
117 beam_stats_to_bins(#beam_stats{}=BeamStats) ->
119 fun (M) -> beam_stats_msg_graphite:add_path_prefix(M, ?PATH_PREFIX) end,
120 Msgs1 = beam_stats_msg_graphite:of_beam_stats(BeamStats),
121 Msgs2 = lists:map(MsgAddPrefix, Msgs1),
122 lists:map(fun beam_stats_msg_graphite:to_bin/1, Msgs2).