X-Git-Url: https://git.xandkar.net/?a=blobdiff_plain;f=src%2Fbeam_stats_msg_graphite.erl;h=159900e0a99bbb53c37a43a530b3fc52c6fadbc9;hb=630be0c8147db5d104fcc5ac6674d2a42d925159;hp=b8d88a14053f5dfb88f3b289427f5efa188b9050;hpb=4bb8ddfe1a69842312c68d25aa3f69db53280921;p=beam_stats.git diff --git a/src/beam_stats_msg_graphite.erl b/src/beam_stats_msg_graphite.erl index b8d88a1..159900e 100644 --- a/src/beam_stats_msg_graphite.erl +++ b/src/beam_stats_msg_graphite.erl @@ -13,14 +13,22 @@ -export( [ of_beam_stats/1 - %, to_bin/1 + , of_beam_stats/2 + , to_iolist/1 + , path_to_iolist/1 + , node_id_to_bin/1 ]). +-define(SCHEMA_VERSION, <<"beam_stats_v0">>). -define(T, #?MODULE). -type t() :: ?T{}. +%% ============================================================================ +%% API +%% ============================================================================ + -spec of_beam_stats(beam_stats:t()) -> [t()]. of_beam_stats(#beam_stats{node_id=NodeID}=BeamStats) -> @@ -33,7 +41,6 @@ of_beam_stats(#beam_stats { timestamp = Timestamp , node_id = _ , memory = Memory - % TODO: Handle the rest of data points , io_bytes_in = IOBytesIn , io_bytes_out = IOBytesOut , context_switches = ContextSwitches @@ -46,15 +53,71 @@ of_beam_stats(#beam_stats ) -> Ts = Timestamp, N = NodeID, - [ cons([N, <<"io">> , <<"bytes_in">> ], IOBytesIn , Ts) - , cons([N, <<"io">> , <<"bytes_out">>], IOBytesOut , Ts) - , cons([N, <<"context_switches">> ], ContextSwitches, Ts) - , cons([N, <<"reductions">> ], Reductions , Ts) - , cons([N, <<"run_queue">> ], RunQueue , Ts) - | of_memory(Memory, NodeID, Ts) - ] - ++ of_ets(ETS, NodeID, Ts) - ++ of_processes(Processes, NodeID, Ts). + Msgs = + [ cons([N, <<"io">> , <<"bytes_in">> ], IOBytesIn , Ts) + , cons([N, <<"io">> , <<"bytes_out">>], IOBytesOut , Ts) + , cons([N, <<"context_switches">> ], ContextSwitches, Ts) + , cons([N, <<"reductions">> ], Reductions , Ts) + , cons([N, <<"run_queue">> ], RunQueue , Ts) + | of_memory(Memory, NodeID, Ts) + ] + ++ of_ets(ETS, NodeID, Ts) + ++ of_processes(Processes, NodeID, Ts), + lists:map(fun path_prefix_schema_version/1, Msgs). + +-spec to_iolist(t()) -> + iolist(). +to_iolist( + ?T + { path = Path + , value = Value + , timestamp = Timestamp + } +) -> + PathIOList = path_to_iolist(Path), + ValueBin = integer_to_binary(Value), + TimestampInt = timestamp_to_integer(Timestamp), + TimestampBin = integer_to_binary(TimestampInt), + [PathIOList, <<" ">>, ValueBin, <<" ">>, TimestampBin, <<"\n">>]. + +-spec path_to_iolist([binary()]) -> + iolist(). +path_to_iolist(Path) -> + list_interleave(Path, <<".">>). + +-spec node_id_to_bin(node()) -> + binary(). +node_id_to_bin(NodeID) -> + NodeIDBin = atom_to_binary(NodeID, utf8), + re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]). + +%% ============================================================================ +%% Helpers +%% ============================================================================ + +-spec path_prefix_schema_version(t()) -> + t(). +path_prefix_schema_version(?T{}=T) -> + path_prefix(T, ?SCHEMA_VERSION). + +-spec path_prefix(t(), binary()) -> + t(). +path_prefix(?T{path=Path}=T, <>) -> + T?T{path = [Prefix | Path]}. + +-spec list_interleave([A], A) -> + [A]. +list_interleave([], _) -> + []; +list_interleave([X], _) -> + [X]; +list_interleave([X|Xs], Sep) -> + [X, Sep | list_interleave(Xs, Sep)]. + +-spec timestamp_to_integer(erlang:timestamp()) -> + non_neg_integer(). +timestamp_to_integer({Megaseconds, Seconds, _}) -> + Megaseconds * 1000000 + Seconds. -spec of_memory([{atom(), non_neg_integer()}], binary(), erlang:timestamp()) -> [t()]. @@ -66,12 +129,13 @@ of_memory(Memory, <>, Timestamp) -> end, lists:map(ComponentToMessage, Memory). --spec of_ets(beam_stats_ets_table:t(), binary(), erlang:timestamp()) -> +-spec of_ets(beam_stats_ets:t(), binary(), erlang:timestamp()) -> [t()]. of_ets(PerTableStats, <>, Timestamp) -> OfEtsTable = fun (Table) -> of_ets_table(Table, NodeID, Timestamp) end, - NestedMsgs = lists:map(OfEtsTable, PerTableStats), - lists:append(NestedMsgs). + MsgsNested = lists:map(OfEtsTable, PerTableStats), + MsgsFlattened = lists:append(MsgsNested), + aggregate_by_path(MsgsFlattened, Timestamp). -spec of_ets_table(beam_stats_ets_table:t(), binary(), erlang:timestamp()) -> [t()]. @@ -84,9 +148,13 @@ of_ets_table(#beam_stats_ets_table <>, Timestamp ) -> - IDBin = beam_stats_ets_table:id_to_bin(ID), + IDType = + case ID =:= Name + of true -> <<"NAMED">> + ; false -> <<"TID">> + end, NameBin = atom_to_binary(Name, latin1), - NameAndID = [NameBin, IDBin], + NameAndID = [NameBin, IDType], [ cons([NodeID, <<"ets_table">>, <<"size">> | NameAndID], Size , Timestamp) , cons([NodeID, <<"ets_table">>, <<"memory">> | NameAndID], Memory, Timestamp) ]. @@ -111,6 +179,7 @@ of_processes( OfProcess = fun (P) -> of_process(P, NodeID, Timestamp) end, PerProcessMsgsNested = lists:map(OfProcess, Processes), PerProcessMsgsFlattened = lists:append(PerProcessMsgsNested), + PerProcessMsgsAggregates = aggregate_by_path(PerProcessMsgsFlattened, Timestamp), Ts = Timestamp, N = NodeID, [ cons([N, <<"processes_count_all">> ], CountAll , Ts) @@ -121,38 +190,50 @@ of_processes( , cons([N, <<"processes_count_running">> ], CountRunning , Ts) , cons([N, <<"processes_count_suspended">> ], CountSuspended , Ts) , cons([N, <<"processes_count_waiting">> ], CountWaiting , Ts) - | PerProcessMsgsFlattened + | PerProcessMsgsAggregates ]. -spec of_process(beam_stats_process:t(), binary(), erlang:timestamp()) -> [t()]. of_process( #beam_stats_process - { pid = Pid + { pid = _ , memory = Memory , total_heap_size = TotalHeapSize , stack_size = StackSize , message_queue_len = MsgQueueLen + , reductions = Reductions }=Process, <>, Timestamp ) -> Origin = beam_stats_process:get_best_known_origin(Process), OriginBin = proc_origin_to_bin(Origin), - PidBin = pid_to_bin(Pid), - OriginAndPid = [OriginBin, PidBin], Ts = Timestamp, N = NodeID, - [ cons([N, <<"process_memory">> , OriginAndPid], Memory , Ts) - , cons([N, <<"process_total_heap_size">> , OriginAndPid], TotalHeapSize , Ts) - , cons([N, <<"process_stack_size">> , OriginAndPid], StackSize , Ts) - , cons([N, <<"process_message_queue_len">> , OriginAndPid], MsgQueueLen , Ts) + [ cons([N, <<"process_memory">> , OriginBin], Memory , Ts) + , cons([N, <<"process_total_heap_size">> , OriginBin], TotalHeapSize , Ts) + , cons([N, <<"process_stack_size">> , OriginBin], StackSize , Ts) + , cons([N, <<"process_message_queue_len">> , OriginBin], MsgQueueLen , Ts) + , cons([N, <<"process_reductions">> , OriginBin], Reductions , Ts) ]. +-spec aggregate_by_path([t()], erlang:timestamp()) -> + [t()]. +aggregate_by_path(Msgs, Timestamp) -> + Aggregate = + fun (?T{path=K, value=V}, ValsByPath) -> + dict:update_counter(K, V, ValsByPath) + end, + ValsByPathDict = lists:foldl(Aggregate, dict:new(), Msgs), + ValsByPathList = dict:to_list(ValsByPathDict), + [cons(Path, Value, Timestamp) || {Path, Value} <- ValsByPathList]. + -spec proc_origin_to_bin(beam_stats_process:best_known_origin()) -> binary(). proc_origin_to_bin({registered_name, Name}) -> - atom_to_binary(Name, utf8); + NameBin = atom_to_binary(Name, utf8), + <<"named--", NameBin/binary>>; proc_origin_to_bin({ancestry, Ancestry}) -> #beam_stats_process_ancestry { raw_initial_call = InitCallRaw @@ -165,7 +246,8 @@ proc_origin_to_bin({ancestry, Ancestry}) -> AncestorsBinOpt = hope_option:map(AncestorsOpt , fun ancestors_to_bin/1), AncestorsBin = hope_option:get(AncestorsBinOpt , Blank), InitCallRawBin = mfa_to_bin(InitCallRaw), - << InitCallRawBin/binary + << "spawned-via--" + , InitCallRawBin/binary , "--" , InitCallOTPBin/binary , "--" @@ -186,12 +268,7 @@ ancestors_to_bin([A | Ancestors]) -> ancestor_to_bin(A) when is_atom(A) -> atom_to_binary(A, utf8); ancestor_to_bin(A) when is_pid(A) -> - pid_to_bin(A). - -pid_to_bin(Pid) -> - PidList = erlang:pid_to_list(Pid), - PidBin = re:replace(PidList, "[\.]", "_", [global, {return, binary}]), - re:replace(PidBin , "[><]", "" , [global, {return, binary}]). + <<"PID">>. -spec mfa_to_bin(mfa()) -> binary(). @@ -209,9 +286,3 @@ cons(Path, Value, Timestamp) -> , value = Value , timestamp = Timestamp }. - --spec node_id_to_bin(node()) -> - binary(). -node_id_to_bin(NodeID) -> - NodeIDBin = atom_to_binary(NodeID, utf8), - re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]).