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