X-Git-Url: https://git.xandkar.net/?a=blobdiff_plain;f=src%2Fbeam_stats_msg_graphite.erl;h=73b770c35d541846de571b0677861c7c25b9026c;hb=e86061d0f13b72a2ea04a96c5c0a6585b58a7beb;hp=62028b1bcafbee8ec78743135c9a5b38c9f4b57f;hpb=9b9299d9a59c8e5236c49c048846c9c4d5ed4a8b;p=beam_stats.git diff --git a/src/beam_stats_msg_graphite.erl b/src/beam_stats_msg_graphite.erl index 62028b1..73b770c 100644 --- a/src/beam_stats_msg_graphite.erl +++ b/src/beam_stats_msg_graphite.erl @@ -13,10 +13,13 @@ -export( [ of_beam_stats/1 - , to_bin/1 - , path_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() :: @@ -38,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 @@ -51,47 +53,66 @@ 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_bin(t()) -> - binary(). -to_bin( +-spec to_iolist(t()) -> + iolist(). +to_iolist( ?T { path = Path , value = Value , timestamp = Timestamp } ) -> - PathBin = path_to_bin(Path), + PathIOList = path_to_iolist(Path), ValueBin = integer_to_binary(Value), TimestampInt = timestamp_to_integer(Timestamp), TimestampBin = integer_to_binary(TimestampInt), - <>. + [PathIOList, <<" ">>, ValueBin, <<" ">>, TimestampBin]. + +-spec path_to_iolist([binary()]) -> + iolist(). +path_to_iolist(Path) -> + list_interleave(Path, <<".">>). --spec path_to_bin([binary()]) -> +-spec node_id_to_bin(node()) -> binary(). -path_to_bin(Path) -> - bin_join(Path, <<".">>). +node_id_to_bin(NodeID) -> + NodeIDBin = atom_to_binary(NodeID, utf8), + re:replace(NodeIDBin, "[\@\.]", "_", [global, {return, binary}]). %% ============================================================================ %% Helpers %% ============================================================================ --spec bin_join([binary()], binary()) -> - binary(). -bin_join([] , <<_/binary>> ) -> <<>>; -bin_join([<> | []] , <<_/binary>> ) -> B; -bin_join([<> | [_|_]=Bins], <>) -> - BinsBin = bin_join(Bins, Sep), - <>. +-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(). @@ -108,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()]. @@ -126,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) ]. @@ -153,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) @@ -163,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 @@ -207,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 , "--" @@ -228,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(). @@ -251,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}]).