src/beam_stats_consumer_statsd.erl

   1 -module(beam_stats_consumer_statsd).
   2
   3 -include("include/beam_stats.hrl").
   4 -include("beam_stats_logging.hrl").
   5
   6 -behaviour(beam_stats_consumer).
   7
   8 -export_type(
   9     [ option/0
  10     ]).
  11
  12 %% Consumer interface
  13 -export(
  14     [ init/1
  15     , consume/2
  16     , terminate/1
  17     ]).
  18
  19 -type option() ::
  20       {consumption_interval , non_neg_integer()}
  21     | {dst_host             , inet:ip_address() | inet:hostname()}
  22     | {dst_port             , inet:port_number()}
  23     | {src_port             , inet:port_number()}
  24     | {num_msgs_per_packet  , non_neg_integer()}
  25     | {static_node_name     , binary()}
  26     .
  27
  28 -define(DEFAULT_DST_HOST, "localhost").
  29 -define(DEFAULT_DST_PORT, 8125).
  30 -define(DEFAULT_SRC_PORT, 8124).
  31
  32 -record(state,
  33     { sock     :: hope_option:t(gen_udp:socket())
  34     , dst_host :: inet:ip_address() | inet:hostname()
  35     , dst_port :: inet:port_number()
  36     , src_port :: inet:port_number()
  37     , num_msgs_per_packet :: non_neg_integer()
  38     , static_node_name    :: hope_option:t(binary())
  39     }).
  40
  41 -type state() ::
  42     #state{}.
  43
  44 %% ============================================================================
  45 %% Consumer implementation
  46 %% ============================================================================
  47
  48 -spec init([option()]) ->
  49     {non_neg_integer(), state()}.
  50 init(Options) ->
  51     ConsumptionInterval = hope_kv_list:get(Options, consumption_interval, 60000),
  52     DstHost = hope_kv_list:get(Options, dst_host, ?DEFAULT_DST_HOST),
  53     DstPort = hope_kv_list:get(Options, dst_port, ?DEFAULT_DST_PORT),
  54     SrcPort = hope_kv_list:get(Options, src_port, ?DEFAULT_SRC_PORT),
  55     NumMsgsPerPacket = hope_kv_list:get(Options, num_msgs_per_packet, 10),
  56     StaticNodeNameOpt = hope_kv_list:get(Options, static_node_name),
  57     State = #state
  58         { sock     = none
  59         , dst_host = DstHost
  60         , dst_port = DstPort
  61         , src_port = SrcPort
  62         , num_msgs_per_packet = NumMsgsPerPacket
  63         , static_node_name    = StaticNodeNameOpt
  64         },
  65     {ConsumptionInterval, State}.
  66
  67 -spec consume(beam_stats_consumer:queue(), state()) ->
  68     state().
  69 consume(
  70     Q,
  71     #state
  72     { num_msgs_per_packet = NumMsgsPerPacket
  73     , static_node_name    = StaticNodeNameOpt
  74     }=State
  75 ) ->
  76     Packets = beam_stats_queue_to_packets(Q, NumMsgsPerPacket, StaticNodeNameOpt),
  77     lists:foldl(fun try_to_connect_and_send/2, State, Packets).
  78
  79 -spec terminate(state()) ->
  80     {}.
  81 terminate(#state{sock=SockOpt}) ->
  82     hope_option:iter(SockOpt, fun gen_udp:close/1).
  83
  84 %% ============================================================================
  85 %% Transport
  86 %% ============================================================================
  87
  88 -spec try_to_connect_and_send(iolist(), state()) ->
  89     state().
  90 try_to_connect_and_send(Payload, #state{}=State1) ->
  91     State2 = try_to_connect_if_no_socket(State1),
  92     try_to_send(State2, Payload).
  93
  94 -spec try_to_send(state(), iolist()) ->
  95     state().
  96 try_to_send(#state{sock=none}=State, _) ->
  97     ?log_error("Sending failed. No socket in state."),
  98     % TODO: Maybe schedule retry?
  99     State;
 100 try_to_send(
 101     #state
 102     { sock     = {some, Sock}
 103     , dst_host = DstHost
 104     , dst_port = DstPort
 105     }=State,
 106     Payload
 107 ) ->
 108     case gen_udp:send(Sock, DstHost, DstPort, Payload)
 109     of  ok ->
 110             State
 111     ;   {error, _}=Error ->
 112             ?log_error(
 113                 "gen_udp:send(~p, ~p, ~p, Payload) -> ~p",
 114                 [Sock, DstHost, DstPort, Error]
 115             ),
 116             % TODO: Do something with unsent messages?
 117             ok = gen_udp:close(Sock),
 118             State#state{sock=none}
 119     end.
 120
 121 -spec try_to_connect_if_no_socket(state()) ->
 122     state().
 123 try_to_connect_if_no_socket(#state{sock={some, _}}=State) ->
 124     State;
 125 try_to_connect_if_no_socket(#state{sock=none, src_port=SrcPort}=State) ->
 126     case gen_udp:open(SrcPort)
 127     of  {ok, Sock} ->
 128             State#state{sock = {some, Sock}}
 129     ;   {error, _}=Error ->
 130             ?log_error("gen_udp:open(~p) -> ~p", [SrcPort, Error]),
 131             State#state{sock = none}
 132     end.
 133
 134 %% ============================================================================
 135 %% Serialization
 136 %% ============================================================================
 137
 138 -spec beam_stats_queue_to_packets(
 139     beam_stats_consumer:queue(),
 140     non_neg_integer(),
 141     hope_option:t(binary())
 142 ) ->
 143     [iolist()].
 144 beam_stats_queue_to_packets(Q, NumMsgsPerPacket, StaticNodeNameOpt) ->
 145     MsgIOLists = lists:append([beam_stats_to_iolists(B, StaticNodeNameOpt) || B <- queue:to_list(Q)]),
 146     hope_list:divide(MsgIOLists, NumMsgsPerPacket).
 147
 148 -spec beam_stats_to_iolists(beam_stats:t(), hope_option:t(binary())) ->
 149     [iolist()].
 150 beam_stats_to_iolists(#beam_stats{node_id=NodeID}=BeamStats, StaticNodeNameOpt) ->
 151     NodeIDBinDefault = beam_stats_msg_graphite:node_id_to_bin(NodeID),
 152     NodeIDBin = hope_option:get(StaticNodeNameOpt, NodeIDBinDefault),
 153     MsgsGraphite = beam_stats_msg_graphite:of_beam_stats(BeamStats, NodeIDBin),
 154     MsgsStatsD =
 155         lists:map(fun beam_stats_msg_statsd_gauge:of_msg_graphite/1, MsgsGraphite),
 156     lists:map(fun beam_stats_msg_statsd_gauge:to_iolist/1, MsgsStatsD).