src/beam_stats_msg_graphite.erl

   1 -module(beam_stats_msg_graphite).
   2
   3 -include("include/beam_stats.hrl").
   4 -include("include/beam_stats_ets_table.hrl").
   5 -include("include/beam_stats_msg_graphite.hrl").
   6 -include("include/beam_stats_process.hrl").
   7 -include("include/beam_stats_process_ancestry.hrl").
   8 -include("include/beam_stats_processes.hrl").
   9
  10 -export_type(
  11     [ t/0
  12     ]).
  13
  14 -export(
  15     [ of_beam_stats/1
  16     , to_bin/1
  17     , path_to_bin/1
  18     ]).
  19
  20 -define(T, #?MODULE).
  21
  22 -type t() ::
  23     ?T{}.
  24
  25 %% ============================================================================
  26 %% API
  27 %% ============================================================================
  28
  29 -spec of_beam_stats(beam_stats:t()) ->
  30     [t()].
  31 of_beam_stats(#beam_stats{node_id=NodeID}=BeamStats) ->
  32     NodeIDBin = node_id_to_bin(NodeID),
  33     of_beam_stats(BeamStats, NodeIDBin).
  34
  35 -spec of_beam_stats(beam_stats:t(), binary()) ->
  36     [t()].
  37 of_beam_stats(#beam_stats
  38     { timestamp = Timestamp
  39     , node_id   = _
  40     , memory    = Memory
  41     % TODO: Handle the rest of data points
  42     , io_bytes_in      = IOBytesIn
  43     , io_bytes_out     = IOBytesOut
  44     , context_switches = ContextSwitches
  45     , reductions       = Reductions
  46     , run_queue        = RunQueue
  47     , ets              = ETS
  48     , processes        = Processes
  49     },
  50     <<NodeID/binary>>
  51 ) ->
  52     Ts = Timestamp,
  53     N = NodeID,
  54     [ cons([N, <<"io">>               , <<"bytes_in">> ], IOBytesIn      , Ts)
  55     , cons([N, <<"io">>               , <<"bytes_out">>], IOBytesOut     , Ts)
  56     , cons([N, <<"context_switches">>                  ], ContextSwitches, Ts)
  57     , cons([N, <<"reductions">>                        ], Reductions     , Ts)
  58     , cons([N, <<"run_queue">>                         ], RunQueue       , Ts)
  59     | of_memory(Memory, NodeID, Ts)
  60     ]
  61     ++ of_ets(ETS, NodeID, Ts)
  62     ++ of_processes(Processes, NodeID, Ts).
  63
  64 -spec to_bin(t()) ->
  65     binary().
  66 to_bin(
  67     ?T
  68     { path      = Path
  69     , value     = Value
  70     , timestamp = Timestamp
  71     }
  72 ) ->
  73     PathBin = path_to_bin(Path),
  74     ValueBin = integer_to_binary(Value),
  75     TimestampInt = timestamp_to_integer(Timestamp),
  76     TimestampBin = integer_to_binary(TimestampInt),
  77     <<PathBin/binary, " ", ValueBin/binary, " ", TimestampBin/binary>>.
  78
  79 -spec path_to_bin([binary()]) ->
  80     binary().
  81 path_to_bin(Path) ->
  82     bin_join(Path, <<".">>).
  83
  84 %% ============================================================================
  85 %% Helpers
  86 %% ============================================================================
  87
  88 -spec bin_join([binary()], binary()) ->
  89     binary().
  90 bin_join([]                         , <<_/binary>>  ) -> <<>>;
  91 bin_join([<<B/binary>> | []]        , <<_/binary>>  ) -> B;
  92 bin_join([<<B/binary>> | [_|_]=Bins], <<Sep/binary>>) ->
  93     BinsBin = bin_join(Bins, Sep),
  94     <<B/binary, Sep/binary, BinsBin/binary>>.
  95
  96 -spec timestamp_to_integer(erlang:timestamp()) ->
  97     non_neg_integer().
  98 timestamp_to_integer({Megaseconds, Seconds, _}) ->
  99     Megaseconds * 1000000 + Seconds.
 100
 101 -spec of_memory([{atom(), non_neg_integer()}], binary(), erlang:timestamp()) ->
 102     [t()].
 103 of_memory(Memory, <<NodeID/binary>>, Timestamp) ->
 104     ComponentToMessage =
 105         fun ({Key, Value}) ->
 106             KeyBin = atom_to_binary(Key, latin1),
 107             cons([NodeID, <<"memory">>, KeyBin], Value, Timestamp)
 108         end,
 109     lists:map(ComponentToMessage, Memory).
 110
 111 -spec of_ets(beam_stats_ets_table:t(), binary(), erlang:timestamp()) ->
 112     [t()].
 113 of_ets(PerTableStats, <<NodeID/binary>>, Timestamp) ->
 114     OfEtsTable = fun (Table) -> of_ets_table(Table, NodeID, Timestamp) end,
 115     NestedMsgs = lists:map(OfEtsTable, PerTableStats),
 116     lists:append(NestedMsgs).
 117
 118 -spec of_ets_table(beam_stats_ets_table:t(), binary(), erlang:timestamp()) ->
 119     [t()].
 120 of_ets_table(#beam_stats_ets_table
 121     { id     = ID
 122     , name   = Name
 123     , size   = Size
 124     , memory = Memory
 125     },
 126     <<NodeID/binary>>,
 127     Timestamp
 128 ) ->
 129     IDBin     = beam_stats_ets_table:id_to_bin(ID),
 130     NameBin   = atom_to_binary(Name, latin1),
 131     NameAndID = [NameBin, IDBin],
 132     [ cons([NodeID, <<"ets_table">>, <<"size">>   | NameAndID], Size  , Timestamp)
 133     , cons([NodeID, <<"ets_table">>, <<"memory">> | NameAndID], Memory, Timestamp)
 134     ].
 135
 136 -spec of_processes(beam_stats_processes:t(), binary(), erlang:timestamp()) ->
 137     [t()].
 138 of_processes(
 139     #beam_stats_processes
 140     { individual_stats         = Processes
 141     , count_all                = CountAll
 142     , count_exiting            = CountExiting
 143     , count_garbage_collecting = CountGarbageCollecting
 144     , count_registered         = CountRegistered
 145     , count_runnable           = CountRunnable
 146     , count_running            = CountRunning
 147     , count_suspended          = CountSuspended
 148     , count_waiting            = CountWaiting
 149     },
 150     <<NodeID/binary>>,
 151     Timestamp
 152 ) ->
 153     OfProcess = fun (P) -> of_process(P, NodeID, Timestamp) end,
 154     PerProcessMsgsNested = lists:map(OfProcess, Processes),
 155     PerProcessMsgsFlattened = lists:append(PerProcessMsgsNested),
 156     Ts = Timestamp,
 157     N  = NodeID,
 158     [ cons([N, <<"processes_count_all">>               ], CountAll              , Ts)
 159     , cons([N, <<"processes_count_exiting">>           ], CountExiting          , Ts)
 160     , cons([N, <<"processes_count_garbage_collecting">>], CountGarbageCollecting, Ts)
 161     , cons([N, <<"processes_count_registered">>        ], CountRegistered       , Ts)
 162     , cons([N, <<"processes_count_runnable">>          ], CountRunnable         , Ts)
 163     , cons([N, <<"processes_count_running">>           ], CountRunning          , Ts)
 164     , cons([N, <<"processes_count_suspended">>         ], CountSuspended        , Ts)
 165     , cons([N, <<"processes_count_waiting">>           ], CountWaiting          , Ts)
 166     | PerProcessMsgsFlattened
 167     ].
 168
 169 -spec of_process(beam_stats_process:t(), binary(), erlang:timestamp()) ->
 170     [t()].
 171 of_process(
 172     #beam_stats_process
 173     { pid               = Pid
 174     , memory            = Memory
 175     , total_heap_size   = TotalHeapSize
 176     , stack_size        = StackSize
 177     , message_queue_len = MsgQueueLen
 178     }=Process,
 179     <<NodeID/binary>>,
 180     Timestamp
 181 ) ->
 182     Origin = beam_stats_process:get_best_known_origin(Process),
 183     OriginBin = proc_origin_to_bin(Origin),
 184     PidBin = pid_to_bin(Pid),
 185     OriginAndPid = [OriginBin, PidBin],
 186     Ts = Timestamp,
 187     N  = NodeID,
 188     [ cons([N, <<"process_memory">>            | OriginAndPid], Memory        , Ts)
 189     , cons([N, <<"process_total_heap_size">>   | OriginAndPid], TotalHeapSize , Ts)
 190     , cons([N, <<"process_stack_size">>        | OriginAndPid], StackSize     , Ts)
 191     , cons([N, <<"process_message_queue_len">> | OriginAndPid], MsgQueueLen   , Ts)
 192     ].
 193
 194 -spec proc_origin_to_bin(beam_stats_process:best_known_origin()) ->
 195     binary().
 196 proc_origin_to_bin({registered_name, Name}) ->
 197     atom_to_binary(Name, utf8);
 198 proc_origin_to_bin({ancestry, Ancestry}) ->
 199     #beam_stats_process_ancestry
 200     { raw_initial_call  = InitCallRaw
 201     , otp_initial_call  = InitCallOTPOpt
 202     , otp_ancestors     = AncestorsOpt
 203     } = Ancestry,
 204     Blank             = <<"NONE">>,
 205     InitCallOTPBinOpt = hope_option:map(InitCallOTPOpt   , fun mfa_to_bin/1),
 206     InitCallOTPBin    = hope_option:get(InitCallOTPBinOpt, Blank),
 207     AncestorsBinOpt   = hope_option:map(AncestorsOpt     , fun ancestors_to_bin/1),
 208     AncestorsBin      = hope_option:get(AncestorsBinOpt  , Blank),
 209     InitCallRawBin    = mfa_to_bin(InitCallRaw),
 210     << InitCallRawBin/binary
 211      , "--"
 212      , InitCallOTPBin/binary
 213      , "--"
 214      , AncestorsBin/binary
 215     >>.
 216
 217 ancestors_to_bin([]) ->
 218     <<>>;
 219 ancestors_to_bin([A | Ancestors]) ->
 220     ABin = ancestor_to_bin(A),
 221     case ancestors_to_bin(Ancestors)
 222     of  <<>> ->
 223             ABin
 224     ;   <<AncestorsBin/binary>> ->
 225             <<ABin/binary, "-", AncestorsBin/binary>>
 226     end.
 227
 228 ancestor_to_bin(A) when is_atom(A) ->
 229     atom_to_binary(A, utf8);
 230 ancestor_to_bin(A) when is_pid(A) ->
 231     pid_to_bin(A).
 232
 233 pid_to_bin(Pid) ->
 234     PidList = erlang:pid_to_list(Pid),
 235     PidBin = re:replace(PidList, "[\.]", "_", [global, {return, binary}]),
 236              re:replace(PidBin , "[><]", "" , [global, {return, binary}]).
 237
 238 -spec mfa_to_bin(mfa()) ->
 239     binary().
 240 mfa_to_bin({Module, Function, Arity}) ->
 241     ModuleBin   = atom_to_binary(Module  , utf8),
 242     FunctionBin = atom_to_binary(Function, utf8),
 243     ArityBin    = erlang:integer_to_binary(Arity),
 244     <<ModuleBin/binary, "-", FunctionBin/binary, "-", ArityBin/binary>>.
 245
 246 -spec cons([binary()], integer(), erlang:timestamp()) ->
 247     t().
 248 cons(Path, Value, Timestamp) ->
 249     ?T
 250     { path      = Path
 251     , value     = Value
 252     , timestamp = Timestamp
 253     }.
 254
 255 -spec node_id_to_bin(node()) ->
 256     binary().
 257 node_id_to_bin(NodeID) ->
 258     NodeIDBin = atom_to_binary(NodeID, utf8),
 259     re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]).