1 -module(beam_stats_msg_graphite).
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").
22 -define(SCHEMA_VERSION, <<"beam_stats_v0">>).
28 %% ============================================================================
30 %% ============================================================================
32 -spec of_beam_stats(beam_stats:t()) ->
34 of_beam_stats(#beam_stats{node_id=NodeID}=BeamStats) ->
35 NodeIDBin = node_id_to_bin(NodeID),
36 of_beam_stats(BeamStats, NodeIDBin).
38 -spec of_beam_stats(beam_stats:t(), binary()) ->
40 of_beam_stats(#beam_stats
41 { timestamp = Timestamp
44 , io_bytes_in = IOBytesIn
45 , io_bytes_out = IOBytesOut
46 , context_switches = ContextSwitches
47 , reductions = Reductions
48 , run_queue = RunQueue
50 , processes = Processes
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)
64 ++ of_ets(ETS, NodeID, Ts)
65 ++ of_processes(Processes, NodeID, Ts),
66 lists:map(fun path_prefix_schema_version/1, Msgs).
68 -spec to_iolist(t()) ->
74 , timestamp = Timestamp
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].
83 -spec path_to_iolist([binary()]) ->
85 path_to_iolist(Path) ->
86 list_interleave(Path, <<".">>).
88 -spec node_id_to_bin(node()) ->
90 node_id_to_bin(NodeID) ->
91 NodeIDBin = atom_to_binary(NodeID, utf8),
92 re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]).
94 %% ============================================================================
96 %% ============================================================================
98 -spec path_prefix_schema_version(t()) ->
100 path_prefix_schema_version(?T{}=T) ->
101 path_prefix(T, ?SCHEMA_VERSION).
103 -spec path_prefix(t(), binary()) ->
105 path_prefix(?T{path=Path}=T, <<Prefix/binary>>) ->
106 T?T{path = [Prefix | Path]}.
108 -spec list_interleave([A], A) ->
110 list_interleave([], _) ->
112 list_interleave([X], _) ->
114 list_interleave([X|Xs], Sep) ->
115 [X, Sep | list_interleave(Xs, Sep)].
117 -spec timestamp_to_integer(erlang:timestamp()) ->
119 timestamp_to_integer({Megaseconds, Seconds, _}) ->
120 Megaseconds * 1000000 + Seconds.
122 -spec of_memory([{atom(), non_neg_integer()}], binary(), erlang:timestamp()) ->
124 of_memory(Memory, <<NodeID/binary>>, Timestamp) ->
126 fun ({Key, Value}) ->
127 KeyBin = atom_to_binary(Key, latin1),
128 cons([NodeID, <<"memory">>, KeyBin], Value, Timestamp)
130 lists:map(ComponentToMessage, Memory).
132 -spec of_ets(beam_stats_ets:t(), binary(), erlang:timestamp()) ->
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).
140 -spec of_ets_table(beam_stats_ets_table:t(), binary(), erlang:timestamp()) ->
142 of_ets_table(#beam_stats_ets_table
153 of true -> <<"NAMED">>
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)
162 -spec of_processes(beam_stats_processes:t(), binary(), erlang:timestamp()) ->
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
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),
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
196 -spec of_process(beam_stats_process:t(), binary(), erlang:timestamp()) ->
202 , total_heap_size = TotalHeapSize
203 , stack_size = StackSize
204 , message_queue_len = MsgQueueLen
205 , reductions = Reductions
210 Origin = beam_stats_process:get_best_known_origin(Process),
211 OriginBin = proc_origin_to_bin(Origin),
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)
221 -spec aggregate_by_path([t()], erlang:timestamp()) ->
223 aggregate_by_path(Msgs, Timestamp) ->
225 fun (?T{path=K, value=V}, ValsByPath) ->
226 dict:update_counter(K, V, ValsByPath)
228 ValsByPathDict = lists:foldl(Aggregate, dict:new(), Msgs),
229 ValsByPathList = dict:to_list(ValsByPathDict),
230 [cons(Path, Value, Timestamp) || {Path, Value} <- ValsByPathList].
232 -spec proc_origin_to_bin(beam_stats_process:best_known_origin()) ->
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
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),
250 , InitCallRawBin/binary
252 , InitCallOTPBin/binary
254 , AncestorsBin/binary
257 ancestors_to_bin([]) ->
259 ancestors_to_bin([A | Ancestors]) ->
260 ABin = ancestor_to_bin(A),
261 case ancestors_to_bin(Ancestors)
264 ; <<AncestorsBin/binary>> ->
265 <<ABin/binary, "-", AncestorsBin/binary>>
268 ancestor_to_bin(A) when is_atom(A) ->
269 atom_to_binary(A, utf8);
270 ancestor_to_bin(A) when is_pid(A) ->
273 -spec mfa_to_bin(mfa()) ->
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>>.
281 -spec cons([binary()], integer(), erlang:timestamp()) ->
283 cons(Path, Value, Timestamp) ->
287 , timestamp = Timestamp