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