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     .
  25
  26 -define(DEFAULT_DST_HOST, "localhost").
  27 -define(DEFAULT_DST_PORT, 8125).
  28 -define(DEFAULT_SRC_PORT, 8124).
  29
  30 -type metric_type() ::
  31     % TODO: Add other metric types
  32     gauge.
  33
  34 -record(statsd_msg,
  35     { name  :: binary()
  36     , value :: non_neg_integer()
  37     , type  :: metric_type()
  38     }).
  39
  40 -type statsd_msg() ::
  41     #statsd_msg{}.
  42
  43 -record(state,
  44     { sock     :: hope_option:t(gen_udp:socket())
  45     , dst_host :: inet:ip_address() | inet:hostname()
  46     , dst_port :: inet:port_number()
  47     , src_port :: inet:port_number()
  48     }).
  49
  50 -type state() ::
  51     #state{}.
  52
  53 -define(PATH_PREFIX, "beam_stats").
  54
  55 %% ============================================================================
  56 %% Consumer implementation
  57 %% ============================================================================
  58
  59 -spec init([option()]) ->
  60     {non_neg_integer(), state()}.
  61 init(Options) ->
  62     ConsumptionInterval = hope_kv_list:get(Options, consumption_interval, 60000),
  63     DstHost = hope_kv_list:get(Options, dst_host, ?DEFAULT_DST_HOST),
  64     DstPort = hope_kv_list:get(Options, dst_port, ?DEFAULT_DST_PORT),
  65     SrcPort = hope_kv_list:get(Options, src_port, ?DEFAULT_SRC_PORT),
  66     State = #state
  67         { sock     = none
  68         , dst_host = DstHost
  69         , dst_port = DstPort
  70         , src_port = SrcPort
  71         },
  72     {ConsumptionInterval, State}.
  73
  74 -spec consume(beam_stats_consumer:queue(), state()) ->
  75     state().
  76 consume(Q, #state{}=State1) ->
  77     Payload = beam_stats_queue_to_binary(Q),
  78     State2 = try_to_connect_if_no_socket(State1),
  79     try_to_send(State2, Payload).
  80
  81 -spec terminate(state()) ->
  82     {}.
  83 terminate(#state{sock=SockOpt}) ->
  84     ok = hope_option:iter(SockOpt, fun gen_udp:close/1),
  85     {}.
  86
  87 %% ============================================================================
  88 %% Transport
  89 %% ============================================================================
  90
  91 -spec try_to_send(state(), binary()) ->
  92     state().
  93 try_to_send(#state{sock=none}=State, _) ->
  94     ?log_error("Sending failed. No socket in state."),
  95     % TODO: Maybe schedule retry?
  96     State;
  97 try_to_send(
  98     #state
  99     { sock     = {some, Sock}
 100     , dst_host = DstHost
 101     , dst_port = DstPort
 102     }=State,
 103     Payload
 104 ) ->
 105     case gen_udp:send(Sock, DstHost, DstPort, Payload)
 106     of  ok ->
 107             State
 108     ;   {error, _}=Error ->
 109             ?log_error(
 110                 "gen_udp:send(~p, ~p, ~p, ~p) -> ~p",
 111                 [Sock, DstHost, DstPort, Error]
 112             ),
 113             % TODO: Do something with unsent messages?
 114             ok = gen_udp:close(Sock),
 115             State#state{sock=none}
 116     end.
 117
 118 -spec try_to_connect_if_no_socket(state()) ->
 119     state().
 120 try_to_connect_if_no_socket(#state{sock={some, _}}=State) ->
 121     State;
 122 try_to_connect_if_no_socket(#state{sock=none, src_port=SrcPort}=State) ->
 123     case gen_udp:open(SrcPort)
 124     of  {ok, Sock} ->
 125             State#state{sock = {some, Sock}}
 126     ;   {error, _}=Error ->
 127             ?log_error("gen_udp:open(~p) -> ~p", [SrcPort, Error]),
 128             State#state{sock = none}
 129     end.
 130
 131 %% ============================================================================
 132 %% Serialization
 133 %% ============================================================================
 134
 135 -spec beam_stats_queue_to_binary(beam_stats_consumer:queue()) ->
 136     binary().
 137 beam_stats_queue_to_binary(Q) ->
 138     iolist_to_binary([beam_stats_to_bins(B) || B <- queue:to_list(Q)]).
 139
 140 -spec beam_stats_to_bins(beam_stats:t()) ->
 141     [binary()].
 142 beam_stats_to_bins(#beam_stats
 143     { node_id = NodeID
 144     , memory  = Memory
 145     , io_bytes_in  = IOBytesIn
 146     , io_bytes_out = IOBytesOut
 147     , context_switches = ContextSwitches
 148     , reductions       = Reductions
 149     }
 150 ) ->
 151     NodeIDBin = node_id_to_bin(NodeID),
 152     Msgs1 =
 153         [ io_bytes_in_to_msg(IOBytesIn)
 154         , io_bytes_out_to_msg(IOBytesOut)
 155         , context_switches_to_msg(ContextSwitches)
 156         , reductions_to_msg(Reductions)
 157         | memory_to_msgs(Memory)
 158         ],
 159     Msgs2 = [statsd_msg_add_name_prefix(M, NodeIDBin) || M <- Msgs1],
 160     [statsd_msg_to_bin(M) || M <- Msgs2].
 161
 162 -spec reductions_to_msg(non_neg_integer()) ->
 163     statsd_msg().
 164 reductions_to_msg(Reductions) ->
 165     #statsd_msg
 166     { name  = <<"reductions">>
 167     , value = Reductions
 168     , type  = gauge
 169     }.
 170
 171 -spec context_switches_to_msg(non_neg_integer()) ->
 172     statsd_msg().
 173 context_switches_to_msg(ContextSwitches) ->
 174     #statsd_msg
 175     { name  = <<"context_switches">>
 176     , value = ContextSwitches
 177     , type  = gauge
 178     }.
 179
 180 -spec io_bytes_in_to_msg(non_neg_integer()) ->
 181     statsd_msg().
 182 io_bytes_in_to_msg(IOBytesIn) ->
 183     #statsd_msg
 184     { name  = <<"io.bytes_in">>
 185     , value = IOBytesIn
 186     , type  = gauge
 187     }.
 188
 189 -spec io_bytes_out_to_msg(non_neg_integer()) ->
 190     statsd_msg().
 191 io_bytes_out_to_msg(IOBytesOut) ->
 192     #statsd_msg
 193     { name  = <<"io.bytes_out">>
 194     , value = IOBytesOut
 195     , type  = gauge
 196     }.
 197
 198 -spec memory_to_msgs([{atom(), non_neg_integer()}]) ->
 199     [statsd_msg()].
 200 memory_to_msgs(Memory) ->
 201     [memory_component_to_statsd_msg(MC) || MC <- Memory].
 202
 203 -spec memory_component_to_statsd_msg({atom(), non_neg_integer()}) ->
 204     statsd_msg().
 205 memory_component_to_statsd_msg({MemType, MemSize}) when MemSize >= 0 ->
 206     MemTypeBin = atom_to_binary(MemType, latin1),
 207     #statsd_msg
 208     { name  = <<"memory.", MemTypeBin/binary>>
 209     , value = MemSize
 210     , type  = gauge
 211     }.
 212
 213 -spec statsd_msg_add_name_prefix(statsd_msg(), binary()) ->
 214     statsd_msg().
 215 statsd_msg_add_name_prefix(#statsd_msg{name=Name1}=Msg, <<NodeID/binary>>) ->
 216     Prefix = <<?PATH_PREFIX, ".", NodeID/binary, ".">>,
 217     Name2 = <<Prefix/binary, Name1/binary>>,
 218     Msg#statsd_msg{name=Name2}.
 219
 220 -spec statsd_msg_to_bin(statsd_msg()) ->
 221     binary().
 222 statsd_msg_to_bin(
 223     #statsd_msg
 224     { name  = <<Name/binary>>
 225     , value = Value
 226     , type  = Type = gauge
 227     }
 228 ) when Value >= 0 ->
 229     TypeBin = metric_type_to_bin(Type),
 230     ValueBin = integer_to_binary(Value),
 231     << Name/binary
 232      , ":"
 233      , ValueBin/binary
 234      , "|"
 235      , TypeBin/binary
 236      , "\n"
 237     >>.
 238
 239 -spec metric_type_to_bin(metric_type()) ->
 240     binary().
 241 metric_type_to_bin(gauge) ->
 242     <<"g">>.
 243
 244 -spec node_id_to_bin(node()) ->
 245     binary().
 246 node_id_to_bin(NodeID) ->
 247     NodeIDBin = atom_to_binary(NodeID, utf8),
 248     re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]).